Add RawValue support for non-converted Payloads

packages/common/src/converter/payload-converter.ts

@@ -100,6 +100,23 @@ export function mapFromPayloads<K extends string, T = unknown>(
   ) as Record<K, T>;
 }
 
+/**
+ * RawValue is a wrapper over a payload.
+ * A payload that belongs to a RawValue is special in that it bypasses user-defined payload converters,
+ * instead using the default payload converter. The payload still undergoes codec conversion.
+ */
+export class RawValue {
+  private readonly _payload: Payload;
+
+  constructor(value: unknown) {
+    this._payload = defaultPayloadConverter.toPayload(value);
+  }
+
+  get payload(): Payload {
+    return this._payload;
+  }
+}
+
 export interface PayloadConverterWithEncoding {
   /**
    * Converts a value to a {@link Payload}.
@@ -143,6 +160,9 @@ export class CompositePayloadConverter implements PayloadConverter {
    * Returns the first successful result, throws {@link ValueError} if there is no converter that can handle the value.
    */
   public toPayload<T>(value: T): Payload {
+    if (value instanceof RawValue) {
+      return value.payload;
+    }
     for (const converter of this.converters) {
       const result = converter.toPayload(value);
       if (result !== undefined) {
@@ -160,6 +180,7 @@ export class CompositePayloadConverter implements PayloadConverter {
     if (payload.metadata === undefined || payload.metadata === null) {
       throw new ValueError('Missing payload metadata');
     }
+
     const encoding = decode(payload.metadata[METADATA_ENCODING_KEY]);
     const converter = this.converterByEncoding.get(encoding);
     if (converter === undefined) {

packages/test/src/test-integration-split-three.ts

@@ -101,7 +101,7 @@ if ('promiseHooks' in v8) {
           {
             file_path: '/packages/test/src/workflows/stack-tracer.ts',
             function_name: 'enhancedStackTracer',
-            line: 32,
+            line: 33,
             column: 35,
             internal_code: false,
           },
@@ -112,7 +112,7 @@ if ('promiseHooks' in v8) {
           {
             file_path: '/packages/test/src/workflows/stack-tracer.ts',
             function_name: 'enhancedStackTracer',
-            line: 32,
+            line: 33,
             column: 35,
             internal_code: false,
           },

packages/test/src/test-integration-workflows.ts

@@ -14,11 +14,13 @@ import {
   ActivityCancellationType,
   ApplicationFailure,
   defineSearchAttributeKey,
+  RawValue,
   SearchAttributePair,
   SearchAttributeType,
   TypedSearchAttributes,
   WorkflowExecutionAlreadyStartedError,
 } from '@temporalio/common';
+import { temporal } from '@temporalio/proto';
 import { signalSchedulingWorkflow } from './activities/helpers';
 import { activityStartedSignal } from './workflows/definitions';
 import * as workflows from './workflows';
@@ -386,7 +388,7 @@ test('Query workflow metadata returns handler descriptions', async (t) => {
 
   await worker.runUntil(async () => {
     const handle = await startWorkflow(queryWorkflowMetadata);
-    const meta = await handle.query(workflow.workflowMetadataQuery);
+    const meta = (await handle.query(workflow.workflowMetadataQuery)) as temporal.api.sdk.v1.IWorkflowMetadata;
     t.is(meta.definition?.type, 'queryWorkflowMetadata');
     const queryDefinitions = meta.definition?.queryDefinitions;
     // Three built-in ones plus dummyQuery1 and dummyQuery2
@@ -1340,6 +1342,31 @@ test('can register search attributes to dev server', async (t) => {
   await env.teardown();
 });
 
+export async function rawValueWorkflow(value: unknown): Promise<RawValue> {
+  const { rawValueActivity } = workflow.proxyActivities({ startToCloseTimeout: '10s' });
+  return await rawValueActivity(new RawValue(value));
+}
+
+test('workflow and activity can receive/return RawValue', async (t) => {
+  const { executeWorkflow, createWorker } = helpers(t);
+  const worker = await createWorker({
+    activities: {
+      async rawValueActivity(value: unknown): Promise<RawValue> {
+        return new RawValue(value);
+      },
+    },
+  });
+
+  await worker.runUntil(async () => {
+    const testValue = 'test';
+    const rawValue = new RawValue(testValue);
+    const res = await executeWorkflow(rawValueWorkflow, {
+      args: [rawValue],
+    });
+    t.deepEqual(res, testValue);
+  });
+});
+
 export async function ChildWorkflowInfo(): Promise<workflow.RootWorkflowInfo | undefined> {
   let blocked = true;
   workflow.setHandler(unblockSignal, () => {

packages/test/src/workflows/stack-tracer.ts

@@ -4,6 +4,7 @@
  */
 import * as wf from '@temporalio/workflow';
 import type { EnhancedStackTrace } from '@temporalio/workflow/lib/interfaces';
+import { defaultPayloadConverter } from '@temporalio/common/lib/converter/payload-converter';
 import type * as activities from '../activities';
 import { unblockOrCancel } from './unblock-or-cancel';
 

packages/workflow/src/internals.ts

@@ -20,6 +20,7 @@ import {
   WorkflowUpdateValidatorType,
   mapFromPayloads,
   fromPayloadsAtIndex,
+  RawValue,
   WorkflowFunctionWithOptions,
   VersioningBehavior,
   WorkflowDefinitionOptions,
@@ -30,7 +31,7 @@ import {
 } from '@temporalio/common/lib/converter/payload-search-attributes';
 import { composeInterceptors } from '@temporalio/common/lib/interceptors';
 import { makeProtoEnumConverters } from '@temporalio/common/lib/internal-workflow';
-import type { coresdk, temporal } from '@temporalio/proto';
+import type { coresdk } from '@temporalio/proto';
 import { alea, RNG } from './alea';
 import { RootCancellationScope } from './cancellation-scope';
 import { UpdateScope } from './update-scope';
@@ -41,7 +42,6 @@ import {
   DefaultSignalHandler,
   StackTraceSDKInfo,
   StackTraceFileSlice,
-  EnhancedStackTrace,
   StackTraceFileLocation,
   WorkflowInfo,
   WorkflowCreateOptionsInternal,
@@ -263,17 +263,19 @@ export class Activator implements ActivationHandler {
       '__stack_trace',
       {
         handler: () => {
-          return this.getStackTraces()
-            .map((s) => s.formatted)
-            .join('\n\n');
+          return new RawValue(
+            this.getStackTraces()
+              .map((s) => s.formatted)
+              .join('\n\n')
+          );
         },
         description: 'Returns a sensible stack trace.',
       },
     ],
     [
       '__enhanced_stack_trace',
       {
-        handler: (): EnhancedStackTrace => {
+        handler: (): RawValue => {
           const { sourceMap } = this;
           const sdk: StackTraceSDKInfo = { name: 'typescript', version: pkg.version };
           const stacks = this.getStackTraces().map(({ structured: locations }) => ({ locations }));
@@ -293,15 +295,15 @@ export class Activator implements ActivationHandler {
               }
             }
           }
-          return { sdk, stacks, sources };
+          return new RawValue({ sdk, stacks, sources });
         },
         description: 'Returns a stack trace annotated with source information.',
       },
     ],
     [
       '__temporal_workflow_metadata',
       {
-        handler: (): temporal.api.sdk.v1.IWorkflowMetadata => {
+        handler: (): RawValue => {
           const workflowType = this.info.workflowType;
           const queryDefinitions = Array.from(this.queryHandlers.entries()).map(([name, value]) => ({
             name,
@@ -315,14 +317,14 @@ export class Activator implements ActivationHandler {
             name,
             description: value.description,
           }));
-          return {
+          return new RawValue({
             definition: {
               type: workflowType,
               queryDefinitions,
               signalDefinitions,
               updateDefinitions,
             },
-          };
+          });
         },
         description: 'Returns metadata associated with this workflow.',
       },

packages/workflow/src/workflow.ts

@@ -22,6 +22,7 @@
   WorkflowReturnType,
   WorkflowUpdateValidatorType,
   SearchAttributeUpdatePair,
+  RawValue,
   compilePriority,
   WorkflowDefinitionOptionsOrGetter,
 } from '@temporalio/common';