diff --git a/packages/client/src/schedule-helpers.ts b/packages/client/src/schedule-helpers.ts index 82f1bde1b6..5d89614dc7 100644 --- a/packages/client/src/schedule-helpers.ts +++ b/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 diff --git a/packages/client/src/schedule-types.ts b/packages/client/src/schedule-types.ts index 0c384b883e..fbc36f3666 100644 --- a/packages/client/src/schedule-types.ts +++ b/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 = { | '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. diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts index 6e93602d4e..b7b5202072 100644 --- a/packages/client/src/workflow-client.ts +++ b/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( - workflowTypeOrFunc: string | T, + workflowTypeOptions: WorkflowTypeOptions, options: WorkflowStartOptions, interceptors: WorkflowClientInterceptor[] ): Promise { - 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>, interceptors: WorkflowClientInterceptor[] ): Promise { - 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> { 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> { 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>( 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, }); } diff --git a/packages/common/src/converter/payload-converter.ts b/packages/common/src/converter/payload-converter.ts index dc7e420972..b1d6ffe614 100644 --- a/packages/common/src/converter/payload-converter.ts +++ b/packages/common/src/converter/payload-converter.ts @@ -1,6 +1,7 @@ import { decode, encode } from '../encoding'; import { PayloadConverterError, ValueError } from '../errors'; import type { Payload } from '../interfaces'; +import { PayloadTypeHints, TypeHint } from '../type-hints'; import type { SerializationContext } from './serialization-context'; import { encodingKeys, encodingTypes, METADATA_ENCODING_KEY } from './types'; @@ -20,12 +21,17 @@ export interface PayloadConverter { * * Should throw {@link ValueError} if unable to convert. */ - toPayload(value: T, context?: SerializationContext): Payload; + toPayload(value: T, context?: SerializationContext, typeHint?: TypeHint): Payload; /** * Converts a {@link Payload} back to a value. */ - fromPayload(payload: Payload, context?: SerializationContext): T; + fromPayload(payload: Payload, context?: SerializationContext, typeHint?: TypeHint): T; + + /** + * Optional method to validate a type hint for the converter + */ + validateTypeHint?: (typeHint: TypeHint) => boolean; } /** @@ -49,13 +55,17 @@ export function toPayloads(converter: PayloadConverter, ...values: unknown[]): P export function toPayloadsWithContext( converter: PayloadConverter, context: SerializationContext | undefined, - values: unknown[] + values: unknown[], + typeHints?: readonly TypeHint[] ): Payload[] | undefined { if (values.length === 0) { return undefined; } + if (typeHints && typeHints.length !== values.length) { + throw new ValueError(`Got ${typeHints.length} type hints for ${values.length} values`); + } - return values.map((value) => converter.toPayload(value, context)); + return values.map((value, index) => converter.toPayload(value, context, typeHints?.[index])); } /** @@ -101,7 +111,8 @@ export function fromPayloadsAtIndex( converter: PayloadConverter, index: number, payloads?: Payload[] | null, - context?: SerializationContext + context?: SerializationContext, + typeHint?: TypeHint ): T { // To make adding arguments a backwards compatible change if (payloads === undefined || payloads === null || index >= payloads.length) { @@ -111,7 +122,7 @@ export function fromPayloadsAtIndex( if (!payload) { return undefined as any; } - return converter.fromPayload(payload, context); + return converter.fromPayload(payload, context, typeHint); } /** @@ -120,12 +131,17 @@ export function fromPayloadsAtIndex( export function arrayFromPayloads( converter: PayloadConverter, payloads?: Payload[] | null, - context?: SerializationContext + context?: SerializationContext, + typeHints?: readonly TypeHint[] ): unknown[] { if (!payloads) { return []; } - return payloads.map((payload: Payload) => converter.fromPayload(payload, context)); + if (typeHints && typeHints.length !== payloads.length) { + throw new ValueError(`Got ${typeHints.length} type hints for ${payloads.length} values`); + } + + return payloads.map((payload: Payload, index) => converter.fromPayload(payload, context, typeHints?.[index])); } export function mapFromPayloads( @@ -172,14 +188,19 @@ export interface PayloadConverterWithEncoding { * @param value The value to convert. Example values include the Workflow args sent from the Client and the values returned by a Workflow or Activity. * @returns The {@link Payload}, or `undefined` if unable to convert. */ - toPayload(value: T, context?: SerializationContext): Payload | undefined; + toPayload(value: T, context?: SerializationContext, typeHint?: TypeHint): Payload | undefined; /** * Converts a {@link Payload} back to a value. */ - fromPayload(payload: Payload, context?: SerializationContext): T; + fromPayload(payload: Payload, context?: SerializationContext, typeHint?: TypeHint): T; readonly encodingType: string; + + /** + * Optional method to validate a type hint for the converter + */ + validateTypeHint?: (typeHint: TypeHint) => boolean; } /** @@ -207,12 +228,12 @@ export class CompositePayloadConverter implements PayloadConverter { * Tries to run `.toPayload(value)` on each converter in the order provided at construction. * Returns the first successful result, throws {@link ValueError} if there is no converter that can handle the value. */ - public toPayload(value: T, context?: SerializationContext): Payload { + public toPayload(value: T, context?: SerializationContext, typeHint?: TypeHint): Payload { if (value instanceof RawValue) { return value.payload; } for (const converter of this.converters) { - const result = converter.toPayload(value, context); + const result = converter.toPayload(value, context, typeHint); if (result !== undefined) { return result; } @@ -224,7 +245,7 @@ export class CompositePayloadConverter implements PayloadConverter { /** * Run {@link PayloadConverterWithEncoding.fromPayload} based on the `encoding` metadata of the {@link Payload}. */ - public fromPayload(payload: Payload, context?: SerializationContext): T { + public fromPayload(payload: Payload, context?: SerializationContext, typeHint?: TypeHint): T { if (payload.metadata === undefined || payload.metadata === null) { throw new ValueError('Missing payload metadata'); } @@ -234,7 +255,7 @@ export class CompositePayloadConverter implements PayloadConverter { if (converter === undefined) { throw new ValueError(`Unknown encoding: ${encoding}`); } - return converter.fromPayload(payload, context); + return converter.fromPayload(payload, context, typeHint); } } diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts index a501b23d02..0f928afa0f 100644 --- a/packages/common/src/index.ts +++ b/packages/common/src/index.ts @@ -29,12 +29,6 @@ export * from './logger'; export * from './priority'; export * from './metrics'; export * from './retry-policy'; -export type { Timestamp, Duration, StringValue } from './time'; -export * from './worker-deployments'; -export * from './workflow-definition-options'; -export * from './workflow-handle'; -export * from './workflow-options'; -export * from './versioning-intent'; export { SearchAttributes, SearchAttributeValue, @@ -44,6 +38,13 @@ export { TypedSearchAttributes, defineSearchAttributeKey, } from './search-attributes'; +export type { Timestamp, Duration, StringValue } from './time'; +export { TypeHint, PayloadTypeHints } from './type-hints'; +export * from './worker-deployments'; +export * from './workflow-definition-options'; +export * from './workflow-handle'; +export * from './workflow-options'; +export * from './versioning-intent'; /** * Encode a UTF-8 string into a Uint8Array diff --git a/packages/common/src/internal-non-workflow/codec-helpers.ts b/packages/common/src/internal-non-workflow/codec-helpers.ts index 9af3302768..8dba67f8c8 100644 --- a/packages/common/src/internal-non-workflow/codec-helpers.ts +++ b/packages/common/src/internal-non-workflow/codec-helpers.ts @@ -13,6 +13,7 @@ import type { LoadedDataConverter } from '../converter/data-converter'; import type { UserMetadata } from '../user-metadata'; import type { SerializationContext } from '../converter/serialization-context'; import type { DecodedPayload, DecodedProtoFailure, EncodedPayload, EncodedProtoFailure } from './codec-types'; +import { TypeHint } from '../type-hints'; /** * Decode through each codec, starting with the last codec. @@ -130,10 +131,11 @@ export async function encodeToPayload( export async function decodeArrayFromPayloads( converter: LoadedDataConverter, payloads?: Payload[] | null, - context?: SerializationContext + context?: SerializationContext, + typeHints?: readonly TypeHint[] ): Promise { const { payloadConverter, payloadCodecs } = converter; - return arrayFromPayloads(payloadConverter, await decodeOptional(payloadCodecs, payloads, context), context); + return arrayFromPayloads(payloadConverter, await decodeOptional(payloadCodecs, payloads, context), context, typeHints); } /** @@ -143,14 +145,16 @@ export async function decodeFromPayloadsAtIndex( converter: LoadedDataConverter, index: number, payloads?: Payload[] | null, - context?: SerializationContext + context?: SerializationContext, + typeHint?: TypeHint ): Promise { const { payloadConverter, payloadCodecs } = converter; return await fromPayloadsAtIndex( payloadConverter, index, await decodeOptional(payloadCodecs, payloads, context), - context + context, + typeHint ); } @@ -195,13 +199,15 @@ export async function encodeToPayloads( export async function encodeToPayloadsWithContext( converter: LoadedDataConverter, context: SerializationContext | undefined, - values: unknown[] + values: unknown[], + typeHints?: readonly TypeHint[] ): Promise { const { payloadConverter, payloadCodecs } = converter; if (values.length === 0) { return undefined; } - const payloads = toPayloadsWithContext(payloadConverter, context, values); + + const payloads = toPayloadsWithContext(payloadConverter, context, values, typeHints); return payloads ? await encode(payloadCodecs, payloads, context) : undefined; } diff --git a/packages/common/src/type-hints.ts b/packages/common/src/type-hints.ts new file mode 100644 index 0000000000..3fdf8dc0c2 --- /dev/null +++ b/packages/common/src/type-hints.ts @@ -0,0 +1,6 @@ +export type TypeHint = unknown; + +export interface PayloadTypeHints { + inputTypes?: readonly TypeHint[]; + outputType?: TypeHint; +} diff --git a/packages/common/src/workflow-definition-options.ts b/packages/common/src/workflow-definition-options.ts index d8e2a3629a..36f53b8d08 100644 --- a/packages/common/src/workflow-definition-options.ts +++ b/packages/common/src/workflow-definition-options.ts @@ -1,5 +1,15 @@ +import { PayloadTypeHints } from './type-hints'; import type { VersioningBehavior } from './worker-deployments'; +/** NEEDS REAL DOCSTRINGS */ +export interface WorkflowDefinitionConfig { + // Options that can be set per workflow execution + // THOMAS - i'd like to rename this to instanceOptions + workflowDefinitionOptions?: WorkflowDefinitionOptionsOrGetter; + // Options that are set once at workflow definition (static) + staticOptions?: WorkflowStaticOptions; +} + /** * Options that can be used when defining a workflow via {@link setWorkflowOptions}. */ @@ -30,6 +40,17 @@ export interface WorkflowDefinitionOptions { failureExceptionTypes?: Array Error>; } +export interface WorkflowStaticOptions { + /** + * Type hints! For workflow input/output. + * + * Validated at compile-time (can we do this?) + * + * @experimental + */ + typeHints?: PayloadTypeHints; +} + type AsyncFunction = (...args: Args) => Promise; export type WorkflowDefinitionOptionsOrGetter = WorkflowDefinitionOptions | (() => WorkflowDefinitionOptions); @@ -38,6 +59,27 @@ export type WorkflowDefinitionOptionsOrGetter = WorkflowDefinitionOptions | (() * @hidden * A workflow function that has been defined with options from {@link WorkflowDefinitionOptions}. */ -export interface WorkflowFunctionWithOptions extends AsyncFunction { - workflowDefinitionOptions: WorkflowDefinitionOptionsOrGetter; +export type WorkflowFunctionWithOptions = AsyncFunction & + Required>; + +/** @internal */ +export type WorkflowFunctionWithStaticOptions = AsyncFunction & + Required>; + +const workflowDefinitionOptionsProperty = 'workflowDefinitionOptions' satisfies keyof WorkflowFunctionWithOptions< + any[], + any +>; +const workflowStaticOptionsProperty = 'staticOptions' satisfies keyof WorkflowFunctionWithStaticOptions; + +/** @internal */ +export function isWorkflowFunctionWithOptions(obj: unknown): obj is WorkflowFunctionWithOptions { + return typeof obj === 'function' && Object.hasOwn(obj, workflowDefinitionOptionsProperty); +} + +/** @internal */ +export function isWorkflowFunctionWithStaticOptions( + obj: unknown +): obj is WorkflowFunctionWithStaticOptions { + return typeof obj === 'function' && Object.hasOwn(obj, workflowStaticOptionsProperty); } diff --git a/packages/common/src/workflow-options.ts b/packages/common/src/workflow-options.ts index 19e27d95ef..234d9175ca 100644 --- a/packages/common/src/workflow-options.ts +++ b/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 { PayloadTypeHints } from './type-hints'; /** * Defines what happens when trying to start a Workflow with the same ID as a *Closed* Workflow. * @@ -212,6 +212,15 @@ export interface BaseWorkflowOptions { * Priority of a workflow */ priority?: Priority; + + /** + * Type hints! For workflow input/output. + * + * Validated at runtime. + * + * @experimental + */ + typeHints?: PayloadTypeHints; } export type WithWorkflowArgs = T & @@ -259,13 +268,36 @@ export interface WorkflowDurationOptions { export type CommonWorkflowOptions = BaseWorkflowOptions & WorkflowDurationOptions; -export function extractWorkflowType( - workflowTypeOrFunc: string | T | WorkflowFunctionWithOptions -): string { - if (typeof workflowTypeOrFunc === 'string') return workflowTypeOrFunc as string; +export interface WorkflowTypeOptions { + type: string; + typeHints?: PayloadTypeHints; +} + +export function extractWorkflowTypeAndConfig( + workflowTypeOrFunc: string | T, + callSiteTypeHints?: PayloadTypeHints +): WorkflowTypeOptions { + if (typeof workflowTypeOrFunc === 'string') { + return { type: workflowTypeOrFunc, typeHints: callSiteTypeHints }; + } 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 (callSiteTypeHints !== undefined) { + throw new TypeError( + 'Workflow type hints cannot be supplied at the call site when using a workflow function. ' + + 'Use defineWorkflowOptions(..., { staticOptions: { typeHints } }) on the workflow function instead, ' + + 'or pass the workflow type as a string.' + ); + } + const definitionTypeHints = isWorkflowFunctionWithStaticOptions(workflowTypeOrFunc) + ? workflowTypeOrFunc.staticOptions.typeHints + : undefined; + return { + type: workflowTypeOrFunc.name, + typeHints: definitionTypeHints, + }; } throw new TypeError( `Invalid workflow type: expected either a string or a function, got '${typeof workflowTypeOrFunc}'` diff --git a/packages/test/src/payload-converters/type-hints.ts b/packages/test/src/payload-converters/type-hints.ts new file mode 100644 index 0000000000..5eefbcae35 --- /dev/null +++ b/packages/test/src/payload-converters/type-hints.ts @@ -0,0 +1,59 @@ +import type { Payload, SerializationContext, TypeHint } from '@temporalio/common'; +import { + BinaryPayloadConverter, + CompositePayloadConverter, + JsonPayloadConverter, + UndefinedPayloadConverter, +} from '@temporalio/common'; + +interface MapperTypeHint { + toIntermediate(value: T): unknown; + fromIntermediate(value: unknown): T; +} + +function isMapperTypeHint(hint: unknown): hint is MapperTypeHint { + return ( + typeof hint === 'object' && + hint !== null && + typeof (hint as any).toIntermediate === 'function' && + typeof (hint as any).fromIntermediate === 'function' + ); +} + +class MapperJsonPayloadConverter extends JsonPayloadConverter { + validateTypeHint(hint: TypeHint): boolean { + return isMapperTypeHint(hint); + } + + toPayload(value: T, context?: SerializationContext, hint?: TypeHint): Payload | undefined { + if (hint === undefined) return super.toPayload(value); + if (!this.validateTypeHint(hint)) return undefined; + + const mapper = hint as MapperTypeHint; + + // Hint handles rich value -> JSON-safe intermediate. + const intermediate = mapper.toIntermediate(value); + + // Existing JSON converter handles JSON-safe intermediate -> Payload. + return super.toPayload(intermediate); + } + + fromPayload(payload: Payload, context?: SerializationContext, hint?: TypeHint): T { + if (hint === undefined) return super.fromPayload(payload); + if (!this.validateTypeHint(hint)) throw new Error('Invalid mapper type hint'); + + const mapper = hint as MapperTypeHint; + + // Existing JSON converter handles Payload -> JSON-safe intermediate. + const intermediate = super.fromPayload(payload); + + // Hint handles JSON-safe intermediate -> rich value. + return mapper.fromIntermediate(intermediate); + } +} + +export const payloadConverter = new CompositePayloadConverter( + new UndefinedPayloadConverter(), + new BinaryPayloadConverter(), + new MapperJsonPayloadConverter() +); diff --git a/packages/test/src/test-type-hints.ts b/packages/test/src/test-type-hints.ts new file mode 100644 index 0000000000..0de733bf06 --- /dev/null +++ b/packages/test/src/test-type-hints.ts @@ -0,0 +1,205 @@ +import { randomUUID } from 'crypto'; +import type { ExecutionContext } from 'ava'; +import { Client, WorkflowFailedError } from '@temporalio/client'; +import { workflowInterceptorModules } from '@temporalio/testing'; +import { bundleWorkflowCode } from '@temporalio/worker'; +import type { TestWorkflowEnvironment } from './helpers'; +import { bundlerOptions } from './helpers'; +import type { Context } from './helpers-integration'; +import { + configurableHelpers, + createTestWorkflowEnvironment, + makeConfigurableEnvironmentTestFn, +} from './helpers-integration'; +import { + parentWorkflowChildDefinition, + parentWorkflowChildDefinitionInvalidCallSiteHints, + Order, + parentWorkflowChildString, + Receipt, + workflowTypeHints, + workflowWithTypeHints, +} from './workflows/type-hints'; + +const converterPath = require.resolve('./payload-converters/type-hints'); +const dataConverter = { payloadConverterPath: converterPath }; + +function assertReceipt(t: ExecutionContext, receipt: Receipt): void { + t.true(receipt instanceof Receipt); + t.is(receipt.summary(), `order-1:12345`); + t.is(typeof receipt.totalCents, 'bigint'); +} + +const test = makeConfigurableEnvironmentTestFn({ + createTestContext: async () => { + const env = await createTestWorkflowEnvironment(); + const workflowBundle = await bundleWorkflowCode({ + ...bundlerOptions, + workflowInterceptorModules: [...workflowInterceptorModules], + workflowsPath: require.resolve('./workflows/type-hints'), + payloadConverterPath: converterPath, + }); + return { env, workflowBundle }; + }, + teardown: async (c) => { + await c.env.teardown(); + }, +}); + +function makeClient(env: TestWorkflowEnvironment): Client { + return new Client({ + connection: env.client.connection, + namespace: env.client.options.namespace, + dataConverter, + }); +} + +test('workflow definition with call-site type hints is invalid', async (t) => { + const client = makeClient(t.context.env); + + await t.throwsAsync( + client.workflow.execute(workflowWithTypeHints, { + workflowId: `wf-${randomUUID()}`, + taskQueue: 'unused', + args: [new Order('order-1', 12345n)], + typeHints: workflowTypeHints, + }), + { + instanceOf: TypeError, + message: /Workflow type hints cannot be supplied at the call site when using a workflow function/, + } + ); +}); + +test('workflow execute uses definition-supplied input and output type hints', 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(workflowWithTypeHints, { + workflowId: `wf-${randomUUID()}`, + taskQueue: h.taskQueue, + args: [new Order('order-1', 12345n)], + }); + + assertReceipt(t, result); + }); +}); + +test('workflow start and result use definition-supplied input and output type hints', 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 handle = await client.workflow.start(workflowWithTypeHints, { + workflowId: `wf-${randomUUID()}`, + taskQueue: h.taskQueue, + args: [new Order('order-1', 12345n)], + }); + + assertReceipt(t, await handle.result()); + }); +}); + +test('workflow execute uses call-site input and output type hints 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('workflowWithTypeHints', { + workflowId: `wf-${randomUUID()}`, + taskQueue: h.taskQueue, + args: [new Order('order-1', 12345n)], + typeHints: workflowTypeHints, + }); + + assertReceipt(t, result); + }); +}); + +test('workflow start and result use call-site input and output type hints 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 handle = await client.workflow.start('workflowWithTypeHints', { + workflowId: `wf-${randomUUID()}`, + taskQueue: h.taskQueue, + args: [new Order('order-1', 12345n)], + typeHints: workflowTypeHints, + }); + + assertReceipt(t, await handle.result()); + }); +}); + +test('same-type continue-as-new reuses definition-supplied input and output type hints', 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(workflowWithTypeHints, { + workflowId: `wf-${randomUUID()}`, + taskQueue: h.taskQueue, + args: [new Order('order-1', 12345n, 1)], + }); + + assertReceipt(t, result); + }); +}); + +test('child workflow uses definition-supplied input and output type hints', 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 hints 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 hints 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(parentWorkflowChildDefinitionInvalidCallSiteHints, { + workflowId: `wf-${randomUUID()}`, + taskQueue: h.taskQueue, + args: [new Order('order-1', 12345n)], + }), + { instanceOf: WorkflowFailedError } + ); + + t.regex(err?.cause?.message ?? '', /Workflow type hints cannot be supplied at the call site/); + }); +}); diff --git a/packages/test/src/workflows/type-hints.ts b/packages/test/src/workflows/type-hints.ts new file mode 100644 index 0000000000..8200fc2059 --- /dev/null +++ b/packages/test/src/workflows/type-hints.ts @@ -0,0 +1,108 @@ +import type { PayloadTypeHints } from '@temporalio/common'; +import { continueAsNew, defineWorkflowOptions, executeChild } from '@temporalio/workflow'; + +export class Order { + constructor( + readonly id: string, + readonly totalCents: bigint, + readonly remainingRuns = 0 + ) {} + + summary(): string { + return `${this.id}:${this.totalCents}:${this.remainingRuns}`; + } +} + +export class Receipt { + constructor( + readonly orderId: string, + readonly totalCents: bigint + ) {} + + summary(): string { + return `${this.orderId}:${this.totalCents}`; + } +} + +export const orderHint = { + toIntermediate(value: Order): unknown { + return { id: value.id, totalCents: value.totalCents.toString(), remainingRuns: value.remainingRuns }; + }, + fromIntermediate(value: any): Order { + return new Order(value.id, BigInt(value.totalCents), value.remainingRuns ?? 0); + }, +}; + +export const receiptHint = { + toIntermediate(value: Receipt): unknown { + return { orderId: value.orderId, totalCents: value.totalCents.toString() }; + }, + fromIntermediate(value: any): Receipt { + return new Receipt(value.orderId, BigInt(value.totalCents)); + }, +}; + +export const workflowTypeHints: PayloadTypeHints = { inputTypes: [orderHint], outputType: receiptHint }; + +function assertOrder(order: Order): void { + if (!(order instanceof Order)) { + throw new Error('Expected Order input'); + } + if (typeof order.totalCents !== 'bigint') { + throw new Error('Expected Order.totalCents to be a bigint'); + } +} + +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(workflowWithTypeHints, { + staticOptions: { typeHints: workflowTypeHints }, +}); +export async function workflowWithTypeHints(order: Order): Promise { + 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 { + assertOrder(order); + const receipt = await executeChild(workflowWithTypeHints, { args: [order] }); + assertReceipt(receipt); + return receipt; +} +defineWorkflowOptions(parentWorkflowChildDefinition, { + staticOptions: { typeHints: workflowTypeHints }, +}); + +export async function parentWorkflowChildString(order: Order): Promise { + assertOrder(order); + const receipt = await executeChild('workflowWithTypeHints', { + args: [order], + typeHints: workflowTypeHints, + }); + assertReceipt(receipt); + return receipt; +} +defineWorkflowOptions(parentWorkflowChildString, { + staticOptions: { typeHints: workflowTypeHints }, +}); + +export async function parentWorkflowChildDefinitionInvalidCallSiteHints(order: Order): Promise { + await executeChild(workflowWithTypeHints, { + args: [order], + typeHints: workflowTypeHints, + }); +} +defineWorkflowOptions(parentWorkflowChildDefinitionInvalidCallSiteHints, { + workflowDefinitionOptions: { failureExceptionTypes: [TypeError] }, + staticOptions: { typeHints: workflowTypeHints }, +}); diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts index 2d8abdbe78..933188f549 100644 --- a/packages/workflow/src/interfaces.ts +++ b/packages/workflow/src/interfaces.ts @@ -16,6 +16,7 @@ import type { VersioningBehavior, InitialVersioningBehavior, SuggestContinueAsNewReason, + PayloadTypeHints, } from '@temporalio/common'; import { SymbolBasedInstanceOfError } from '@temporalio/common/lib/type-helpers'; import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow/enums-helpers'; @@ -364,6 +365,9 @@ export interface ContinueAsNewOptions { * @experimental Versioning semantics with continue-as-new are experimental and may change in the future. */ initialVersioningBehavior?: InitialVersioningBehavior; + + /** typeHints input in case you CAN to a different workflow */ + typeHints?: PayloadTypeHints } /** diff --git a/packages/workflow/src/internals.ts b/packages/workflow/src/internals.ts index f2e60516ea..8392e5e959 100644 --- a/packages/workflow/src/internals.ts +++ b/packages/workflow/src/internals.ts @@ -15,6 +15,8 @@ import type { VersioningBehavior, WorkflowDefinitionOptions, WorkflowSerializationContext, + PayloadTypeHints, + TypeHint, } from '@temporalio/common'; import { defaultFailureConverter, @@ -114,6 +116,7 @@ export interface Completion { resolve(val: Success): void; reject(reason: Error): void; context?: Context; + typeHint?: TypeHint; } export interface Condition { @@ -502,6 +505,9 @@ export class Activator implements ActivationHandler { protected readonly stackTracesEnabled: boolean; + // THOMAS - maybe store static options directly? + public typeHints?: PayloadTypeHints; + constructor({ info, now, @@ -639,7 +645,7 @@ export class Activator implements ActivationHandler { executeWithLifecycleLogging(() => execute({ headers: activation.headers ?? {}, - args: arrayFromPayloads(this.payloadConverter, activation.arguments, context), + args: arrayFromPayloads(this.payloadConverter, activation.arguments, context, this.typeHints?.inputTypes), }) ).then(this.completeWorkflow.bind(this), this.handleWorkflowFailure.bind(this)) ); @@ -655,7 +661,7 @@ export class Activator implements ActivationHandler { searchAttributes: decodeSearchAttributes(searchAttributes?.indexedFields), typedSearchAttributes: decodeTypedSearchAttributes(searchAttributes?.indexedFields), - + // THOMAS - this is where we'd decode memo with type hints on workflow start memo: mapFromPayloads(this.payloadConverter, memo?.fields, context), lastResult: fromPayloadsAtIndex(this.payloadConverter, 0, lastCompletionResult?.payloads, context), lastFailure: @@ -745,10 +751,10 @@ 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, typeHint } = 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 ? this.payloadConverter.fromPayload(completed.result, context, typeHint) : undefined; resolve(result); } else if (activation.result.failed) { const { failure } = activation.result.failed; @@ -1417,7 +1423,7 @@ export class Activator implements ActivationHandler { this.pushCommand( { completeWorkflowExecution: { - result: this.payloadConverter.toPayload(result, context), + result: this.payloadConverter.toPayload(result, context, this.typeHints?.outputType), }, }, true diff --git a/packages/workflow/src/worker-interface.ts b/packages/workflow/src/worker-interface.ts index eb46119ec7..8b7e30d03c 100644 --- a/packages/workflow/src/worker-interface.ts +++ b/packages/workflow/src/worker-interface.ts @@ -4,7 +4,12 @@ * @module */ import type { WorkflowFunctionWithOptions } from '@temporalio/common'; -import { encodeVersioningBehavior, IllegalStateError } from '@temporalio/common'; +import { + encodeVersioningBehavior, + IllegalStateError, + isWorkflowFunctionWithOptions, + isWorkflowFunctionWithStaticOptions, +} from '@temporalio/common'; import type { coresdk } from '@temporalio/proto'; import type { WorkflowInterceptorsFactory } from './interceptors'; import type { WorkflowCreateOptionsInternal } from './interfaces'; @@ -97,6 +102,9 @@ export function initRuntime(options: WorkflowCreateOptionsInternal): void { activator.workflowDefinitionOptionsGetter = activator.workflow.workflowDefinitionOptions; } } + if (isWorkflowFunctionWithStaticOptions(activator.workflow)) { + activator.typeHints = activator.workflow.staticOptions.typeHints; + } } catch (e) { try { // Core won't send an eviction job after an early error, so we are responsible for triggering @@ -315,8 +323,3 @@ export function destroy(): void { } } } - -function isWorkflowFunctionWithOptions(obj: any): obj is WorkflowFunctionWithOptions { - if (obj == null) return false; - return Object.hasOwn(obj, 'workflowDefinitionOptions'); -} diff --git a/packages/workflow/src/workflow.ts b/packages/workflow/src/workflow.ts index fbf8f97791..e2f8bf58f6 100644 --- a/packages/workflow/src/workflow.ts +++ b/packages/workflow/src/workflow.ts @@ -17,13 +17,14 @@ import type { WorkflowUpdateValidatorType, SearchAttributeUpdatePair, WorkflowDefinitionOptionsOrGetter, + WorkflowDefinitionConfig, } from '@temporalio/common'; import { compileRetryPolicy, compilePriority, encodeActivityCancellationType, encodeWorkflowIdReusePolicy, - extractWorkflowType, + extractWorkflowTypeAndConfig, HandlerUnfinishedPolicy, mapToPayloads, encodeInitialVersioningBehavior, @@ -434,7 +435,7 @@ function startChildWorkflowExecutionNextHandler({ seq, workflowId, workflowType, - input: toPayloadsWithContext(activator.payloadConverter, context, options.args), + input: toPayloadsWithContext(activator.payloadConverter, context, options.args, options.typeHints?.inputTypes), retryPolicy: options.retry ? compileRetryPolicy(options.retry) : undefined, taskQueue: options.taskQueue || activator.info.taskQueue, workflowExecutionTimeout: msOptionalToTs(options.workflowExecutionTimeout), @@ -450,6 +451,7 @@ function startChildWorkflowExecutionNextHandler({ options.searchAttributes || options.typedSearchAttributes ? { indexedFields: encodeUnifiedSearchAttributes(options.searchAttributes, options.typedSearchAttributes) } : undefined, + // THOMAS - memo type hints not supported yet! memo: options.memo && mapToPayloads(activator.payloadConverter, options.memo, context), versioningIntent: versioningIntentToProto(options.versioningIntent), priority: options.priority ? compilePriority(options.priority) : undefined, @@ -477,6 +479,7 @@ function startChildWorkflowExecutionNextHandler({ resolve, reject, context, + typeHint: options.typeHints?.outputType, }); }); untrackPromise(startPromise); @@ -874,7 +877,14 @@ export async function startChild( '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, typeHints } = extractWorkflowTypeAndConfig( + workflowTypeOrFunc, + optionsWithDefaults.typeHints + ); + const workflowOptions = { + ...optionsWithDefaults, + typeHints, + }; const execute = composeInterceptors( activator.interceptors.outbound, 'startChildWorkflowExecution', @@ -882,14 +892,14 @@ export async function startChild( ); 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> { return (await completed) as any; @@ -905,7 +915,7 @@ export async function startChild( args, target: { type: 'child', - childWorkflowId: optionsWithDefaults.workflowId, + childWorkflowId: workflowOptions.workflowId, }, headers: {}, }); @@ -975,7 +985,14 @@ export async function executeChild( '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, typeHints } = extractWorkflowTypeAndConfig( + workflowTypeOrFunc, + optionsWithDefaults.typeHints + ); + const workflowOptions = { + ...optionsWithDefaults, + typeHints, + }; const execute = composeInterceptors( activator.interceptors.outbound, 'startChildWorkflowExecution', @@ -983,7 +1000,7 @@ export async function executeChild( ); const execPromise = execute({ seq: activator.nextSeqs.childWorkflow++, - options: optionsWithDefaults, + options: workflowOptions, headers: {}, workflowType, }); @@ -1067,15 +1084,22 @@ export function makeContinueAsNewFunc( ...rest, }; + let resolvedTypeHints = options?.typeHints; + // If no hints were provided, reuse current workflow hints only for same-type continue-as-new. + if (resolvedTypeHints == null && requiredOptions.workflowType === info.workflowType) { + resolvedTypeHints = activator.typeHints; + } + return (...args: Parameters): Promise => { 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, resolvedTypeHints?.inputTypes), headers, taskQueue: options.taskQueue, + // THOMAS - memo type hints not support yet memo: options.memo && mapToPayloads(activator.payloadConverter, options.memo, context), searchAttributes: options.searchAttributes || options.typedSearchAttributes @@ -1764,6 +1788,9 @@ export function allHandlersFinished(): boolean { /** * Can be used to alter workflow functions with certain options specified at definition time. * + * Prefer {@link defineWorkflowOptions} for new code, especially when setting static workflow + * metadata such as type hints. + * * @example * For example: * ```ts @@ -1799,9 +1826,15 @@ export function setWorkflowOptions( options: WorkflowDefinitionOptionsOrGetter, fn: (...args: A) => Promise ): void { - Object.assign(fn, { - workflowDefinitionOptions: options, - }); + return defineWorkflowOptions(fn, { workflowDefinitionOptions: options }); +} + +/** NEEDS DOCSTRING */ +export function defineWorkflowOptions( + fn: (...args: A) => Promise, + config: WorkflowDefinitionConfig +): void { + Object.assign(fn, config); } export const stackTraceQuery = defineQuery('__stack_trace');