feat: add serialization context to data conversion interfaces

packages/client/src/async-completion-client.ts

@@ -1,5 +1,6 @@
 import { status as grpcStatus } from '@grpc/grpc-js';
-import { ensureTemporalFailure } from '@temporalio/common';
+import { ensureTemporalFailure, LoadedDataConverter, SerializationContext } from '@temporalio/common';
+import { withSerializationContext } from '@temporalio/common/lib/converter/serialization-context';
 import { encodeErrorToFailure, encodeToPayloads } from '@temporalio/common/lib/internal-non-workflow';
 import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { SymbolBasedInstanceOfError } from '@temporalio/common/lib/type-helpers';
@@ -81,6 +82,7 @@ export interface FullActivityId {
  */
 export class AsyncCompletionClient extends BaseClient {
   public readonly options: LoadedAsyncCompletionClientOptions;
+  private context?: SerializationContext;
 
   constructor(options?: AsyncCompletionClientOptions) {
     super(options);
@@ -91,6 +93,13 @@ export class AsyncCompletionClient extends BaseClient {
     };
   }
 
+  /**
+   * Return the data converter bound to the current serialization context, if any.
+   */
+  private get contextDataConverter(): LoadedDataConverter {
+    return this.context ? withSerializationContext(this.dataConverter, this.context) : this.dataConverter;
+  }
+
   /**
    * Raw gRPC access to the Temporal service.
    *
@@ -101,6 +110,20 @@ export class AsyncCompletionClient extends BaseClient {
     return this.connection.workflowService;
   }
 
+  /**
+   * Return a client instance with all serialization operations bound to `context`.
+   */
+  withContext(context: SerializationContext): AsyncCompletionClient {
+    const client = new AsyncCompletionClient({
+      connection: this.connection,
+      dataConverter: this.dataConverter,
+      identity: this.options.identity,
+      namespace: this.options.namespace,
+    });
+    client.context = context;
+    return client;
+  }
+
   /**
    * Transforms grpc errors into well defined TS errors.
    */
@@ -127,7 +150,8 @@ export class AsyncCompletionClient extends BaseClient {
   async complete(fullActivityId: FullActivityId, result: unknown): Promise<void>;
 
   async complete(taskTokenOrFullActivityId: Uint8Array | FullActivityId, result: unknown): Promise<void> {
-    const payloads = await encodeToPayloads(this.dataConverter, result);
+    const dataConverter = this.contextDataConverter;
+    const payloads = await encodeToPayloads(dataConverter, result);
     try {
       if (taskTokenOrFullActivityId instanceof Uint8Array) {
         await this.workflowService.respondActivityTaskCompleted({
@@ -159,7 +183,8 @@ export class AsyncCompletionClient extends BaseClient {
   async fail(fullActivityId: FullActivityId, err: unknown): Promise<void>;
 
   async fail(taskTokenOrFullActivityId: Uint8Array | FullActivityId, err: unknown): Promise<void> {
-    const failure = await encodeErrorToFailure(this.dataConverter, ensureTemporalFailure(err));
+    const dataConverter = this.contextDataConverter;
+    const failure = await encodeErrorToFailure(dataConverter, ensureTemporalFailure(err));
     try {
       if (taskTokenOrFullActivityId instanceof Uint8Array) {
         await this.workflowService.respondActivityTaskFailed({
@@ -191,7 +216,8 @@ export class AsyncCompletionClient extends BaseClient {
   reportCancellation(fullActivityId: FullActivityId, details?: unknown): Promise<void>;
 
   async reportCancellation(taskTokenOrFullActivityId: Uint8Array | FullActivityId, details?: unknown): Promise<void> {
-    const payloads = await encodeToPayloads(this.dataConverter, details);
+    const dataConverter = this.contextDataConverter;
+    const payloads = await encodeToPayloads(dataConverter, details);
     try {
       if (taskTokenOrFullActivityId instanceof Uint8Array) {
         await this.workflowService.respondActivityTaskCanceled({
@@ -223,7 +249,8 @@ export class AsyncCompletionClient extends BaseClient {
   heartbeat(fullActivityId: FullActivityId, details?: unknown): Promise<void>;
 
   async heartbeat(taskTokenOrFullActivityId: Uint8Array | FullActivityId, details?: unknown): Promise<void> {
-    const payloads = await encodeToPayloads(this.dataConverter, details);
+    const dataConverter = this.contextDataConverter;
+    const payloads = await encodeToPayloads(dataConverter, details);
     let cancelRequested = false;
     let paused = false;
     let reset = false;

packages/client/src/schedule-client.ts

@@ -234,7 +234,7 @@ export class ScheduleClient extends BaseClient {
       scheduleId: opts.scheduleId,
       schedule: {
         spec: encodeScheduleSpec(opts.spec),
-        action: await encodeScheduleAction(this.dataConverter, opts.action, headers),
+        action: await encodeScheduleAction(this.dataConverter, this.options.namespace, opts.action, headers),
         policies: encodeSchedulePolicies(opts.policies),
         state: encodeScheduleState(opts.state),
       },
@@ -298,7 +298,7 @@ export class ScheduleClient extends BaseClient {
       scheduleId,
       schedule: {
         spec: encodeScheduleSpec(opts.spec),
-        action: await encodeScheduleAction(this.dataConverter, opts.action, header),
+        action: await encodeScheduleAction(this.dataConverter, this.options.namespace, opts.action, header),
         policies: encodeSchedulePolicies(opts.policies),
         state: encodeScheduleState(opts.state),
       },
@@ -432,7 +432,11 @@ export class ScheduleClient extends BaseClient {
         return {
           scheduleId,
           spec: decodeScheduleSpec(raw.schedule.spec),
-          action: await decodeScheduleAction(this.client.dataConverter, raw.schedule.action),
+          action: await decodeScheduleAction(
+            this.client.dataConverter,
+            this.client.options.namespace,
+            raw.schedule.action
+          ),
           memo: await decodeMapFromPayloads(this.client.dataConverter, raw.memo?.fields),
           searchAttributes: decodeSearchAttributes(raw.searchAttributes?.indexedFields),
           typedSearchAttributes: decodeTypedSearchAttributes(raw.searchAttributes?.indexedFields),

packages/client/src/schedule-helpers.ts

@@ -1,12 +1,14 @@
 import Long from 'long';
 import {
+  WorkflowSerializationContext,
   compilePriority,
   compileRetryPolicy,
   decodePriority,
   decompileRetryPolicy,
   extractWorkflowType,
   LoadedDataConverter,
 } from '@temporalio/common';
+import { withSerializationContext } from '@temporalio/common/lib/converter/serialization-context';
 import { encodeUserMetadata, decodeUserMetadata } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
 import {
   encodeUnifiedSearchAttributes,
@@ -247,16 +249,24 @@ export function encodeScheduleSpec(spec: ScheduleSpec): temporal.api.schedule.v1
 
 export async function encodeScheduleAction(
   dataConverter: LoadedDataConverter,
+  namespace: string,
   action: CompiledScheduleAction,
   headers: Headers
 ): Promise<temporal.api.schedule.v1.IScheduleAction> {
+  const context: WorkflowSerializationContext = {
+    type: 'workflow',
+    namespace,
+    workflowId: action.workflowId,
+    workflowType: action.workflowType,
+  };
+  const contextDataConverter = withSerializationContext(dataConverter, context);
   return {
     startWorkflow: {
       workflowId: action.workflowId,
       workflowType: {
         name: action.workflowType,
       },
-      input: { payloads: await encodeToPayloads(dataConverter, ...action.args) },
+      input: { payloads: await encodeToPayloads(contextDataConverter, ...action.args) },
       taskQueue: {
         kind: temporal.api.enums.v1.TaskQueueKind.TASK_QUEUE_KIND_NORMAL,
         name: action.taskQueue,
@@ -265,15 +275,15 @@ export async function encodeScheduleAction(
       workflowRunTimeout: msOptionalToTs(action.workflowRunTimeout),
       workflowTaskTimeout: msOptionalToTs(action.workflowTaskTimeout),
       retryPolicy: action.retry ? compileRetryPolicy(action.retry) : undefined,
-      memo: action.memo ? { fields: await encodeMapToPayloads(dataConverter, action.memo) } : undefined,
+      memo: action.memo ? { fields: await encodeMapToPayloads(contextDataConverter, action.memo) } : undefined,
       searchAttributes:
         action.searchAttributes || action.typedSearchAttributes // eslint-disable-line @typescript-eslint/no-deprecated
           ? {
               indexedFields: encodeUnifiedSearchAttributes(action.searchAttributes, action.typedSearchAttributes), // eslint-disable-line @typescript-eslint/no-deprecated
             }
           : undefined,
       header: { fields: headers },
-      userMetadata: await encodeUserMetadata(dataConverter, action.staticSummary, action.staticDetails),
+      userMetadata: await encodeUserMetadata(contextDataConverter, action.staticSummary, action.staticDetails),
       priority: action.priority ? compilePriority(action.priority) : undefined,
     },
   };
@@ -321,20 +331,32 @@ export function decodeScheduleSpec(pb: temporal.api.schedule.v1.IScheduleSpec):
 
 export async function decodeScheduleAction(
   dataConverter: LoadedDataConverter,
+  namespace: string,
   pb: temporal.api.schedule.v1.IScheduleAction
 ): Promise<ScheduleDescriptionAction> {
   if (pb.startWorkflow) {
-    const { staticSummary, staticDetails } = await decodeUserMetadata(dataConverter, pb.startWorkflow?.userMetadata);
+    const workflowId = pb.startWorkflow.workflowId!;
+    const workflowType = pb.startWorkflow.workflowType?.name ?? undefined;
+    const contextDataConverter = withSerializationContext(dataConverter, {
+      type: 'workflow',
+      namespace,
+      workflowId,
+      workflowType,
+    });
+    const { staticSummary, staticDetails } = await decodeUserMetadata(
+      contextDataConverter,
+      pb.startWorkflow?.userMetadata
+    );
     return {
       type: 'startWorkflow',
 
-      workflowId: pb.startWorkflow.workflowId!,
+      workflowId,
 
       workflowType: pb.startWorkflow.workflowType!.name!,
 
       taskQueue: pb.startWorkflow.taskQueue!.name!,
-      args: await decodeArrayFromPayloads(dataConverter, pb.startWorkflow.input?.payloads),
-      memo: await decodeMapFromPayloads(dataConverter, pb.startWorkflow.memo?.fields),
+      args: await decodeArrayFromPayloads(contextDataConverter, pb.startWorkflow.input?.payloads),
+      memo: await decodeMapFromPayloads(contextDataConverter, pb.startWorkflow.memo?.fields),
       retry: decompileRetryPolicy(pb.startWorkflow.retryPolicy),
       searchAttributes: decodeSearchAttributes(pb.startWorkflow.searchAttributes?.indexedFields),
       typedSearchAttributes: decodeTypedSearchAttributes(pb.startWorkflow.searchAttributes?.indexedFields),