From d1beb6c5a76db6abb132998ab3d5982926db77a6 Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Wed, 15 Jul 2026 13:46:07 -0400 Subject: [PATCH] feat(proto): auto-discover payload-visitor roots from coresdk Replace the hand-maintained ROOTS list in gen-payload-visitor.ts with automatic discovery: a root is any payload-bearing message under the coresdk namespace that no other reachable message embeds (WorkflowActivation, ActivityTaskCompletion, NexusTask, etc.). Regenerating now picks up new worker-boundary message types without editing the generator. Regenerated payload-visitor.generated.ts. --- packages/proto/scripts/gen-payload-visitor.ts | 62 ++-- .../proto/src/payload-visitor.generated.ts | 284 ++++++++++++++++++ 2 files changed, 329 insertions(+), 17 deletions(-) diff --git a/packages/proto/scripts/gen-payload-visitor.ts b/packages/proto/scripts/gen-payload-visitor.ts index b4b4883c6..4cce764a7 100644 --- a/packages/proto/scripts/gen-payload-visitor.ts +++ b/packages/proto/scripts/gen-payload-visitor.ts @@ -3,18 +3,16 @@ import { writeFileSync } from 'node:fs'; import * as protobuf from 'protobufjs'; import * as prettier from 'prettier'; -// Generates `packages/proto/src/payload-visitor.generated.ts`: a synchronous walk of the -// WorkflowActivation / WorkflowActivationCompletion message trees that invokes a payload visitor at -// every Payload-bearing field, threading a per-message context. Runs in proto's build (and via -// `pnpm gen:payload-visitor`). +// Generates `packages/proto/src/payload-visitor.generated.ts`: a synchronous walk of the worker +// boundary message trees (WorkflowActivation, ActivityTask, NexusTaskCompletion, ...) that invokes a +// payload visitor at every Payload-bearing field, threading a per-message context. Runs in proto's +// build (and via `pnpm gen:payload-visitor`). const PAYLOAD = 'temporal.api.common.v1.Payload'; const ANY = 'google.protobuf.Any'; -const ROOTS = [ - { type: 'coresdk.workflow_activation.WorkflowActivation', entry: 'walkWorkflowActivation' }, - { type: 'coresdk.workflow_completion.WorkflowActivationCompletion', entry: 'walkWorkflowActivationCompletion' }, -] as const; +// Namespaces scanned for root messages. +const ROOT_NAMESPACES = ['coresdk'] as const; const jsonModule = require(resolve(__dirname, '../protos/json-module.js')); const root = protobuf.Root.fromJSON(jsonModule); @@ -40,7 +38,23 @@ function tsType(type: protobuf.Type): string { return segments.join('.'); } -// All message types reachable from the roots, excluding the terminal Payload and the opaque Any. +/** Collect every message type declared (recursively) under a namespace. */ +function collectMessages(obj: protobuf.ReflectionObject, out: protobuf.Type[]): void { + if (obj instanceof protobuf.Type) out.push(obj); + if (obj instanceof protobuf.Namespace) { + for (const child of obj.nestedArray) collectMessages(child, out); + } +} + +// Candidate roots: every message declared under the scanned namespaces. +const candidates: protobuf.Type[] = []; +for (const name of ROOT_NAMESPACES) { + const ns = root.lookup(name); + if (!ns) throw new Error(`Root namespace not found: ${name}`); + collectMessages(ns, candidates); +} + +// All message types reachable from the candidates, excluding the terminal Payload and the opaque Any. const reachableTypes = (): Map => { const types = new Map(); const discover = (type: protobuf.Type): void => { @@ -52,15 +66,14 @@ const reachableTypes = (): Map => { if (message) discover(message); } }; - for (const { type } of ROOTS) discover(root.lookupType(type)); + for (const candidate of candidates) discover(candidate); return types; }; const types = reachableTypes(); -// Which of those types can contain a Payload, directly or nested inside another message? We work -// backwards from Payload. First index the reverse references: `referrers.get(X)` is every type -// that has a field of type X. +// Reverse reference index: `referrers.get(X)` is every type that has a field of type X. A type absent +// from this map is embedded by nothing — that is the "is a root" test below. const referrers = new Map(); for (const type of types.values()) { for (const field of type.fieldsArray) { @@ -87,6 +100,20 @@ while (toVisit.length > 0) { // A message contains a Payload if it is one, or if it was marked above. const hasPayload = (type: protobuf.Type): boolean => fqn(type) === PAYLOAD || reachesPayload.has(fqn(type)); +// Roots: candidate messages that carry a payload and are embedded by no other message. +const roots = candidates.filter((type) => hasPayload(type) && !referrers.has(fqn(type))).sort(byFqn); + +// Public entry-point name for a root, derived from its message name (e.g. `walkWorkflowActivation`). +const entryName = (type: protobuf.Type): string => `walk${type.name}`; + +// Guard: two roots that share a simple name would collide as exported entry points. +const seenEntries = new Set(); +for (const rootType of roots) { + const name = entryName(rootType); + if (seenEntries.has(name)) throw new Error(`Duplicate root entry name '${name}' (from ${fqn(rootType)})`); + seenEntries.add(name); +} + /** Every reachable message type that needs a walker (i.e. can contain a Payload), sorted by name. */ const collectWalkers = (): protobuf.Type[] => [...types.values()].filter(hasPayload).sort(byFqn); @@ -150,12 +177,13 @@ function emit(): string { const walkers = collectWalkers(); const namespaces = [...new Set(walkers.map(topLevelNamespace))].sort(); - const entries = ROOTS.map(({ type, entry }) => { - const t = root.lookupType(type); + const entries = roots.map((type) => { return [ - `export function ${entry}(root: ${tsType(t)}, env: WalkEnv, context: Ctx): Promise[] {`, + `export function ${entryName(type)}(root: ${tsType( + type + )}, env: WalkEnv, context: Ctx): Promise[] {`, ` const pending: Promise[] = [];`, - ` ${fnName(t)}(root, env, context, pending);`, + ` ${fnName(type)}(root, env, context, pending);`, ` return pending;`, `}`, ].join('\n'); diff --git a/packages/proto/src/payload-visitor.generated.ts b/packages/proto/src/payload-visitor.generated.ts index 4df45b61e..dcc8566f0 100644 --- a/packages/proto/src/payload-visitor.generated.ts +++ b/packages/proto/src/payload-visitor.generated.ts @@ -12,6 +12,56 @@ export interface WalkEnv { skipSearchAttributes: boolean; } +export function walkActivityTask( + root: coresdk.activity_task.IActivityTask, + env: WalkEnv, + context: Ctx +): Promise[] { + const pending: Promise[] = []; + walk_coresdk_activity_task_ActivityTask(root, env, context, pending); + return pending; +} + +export function walkActivityHeartbeat( + root: coresdk.IActivityHeartbeat, + env: WalkEnv, + context: Ctx +): Promise[] { + const pending: Promise[] = []; + walk_coresdk_ActivityHeartbeat(root, env, context, pending); + return pending; +} + +export function walkActivityTaskCompletion( + root: coresdk.IActivityTaskCompletion, + env: WalkEnv, + context: Ctx +): Promise[] { + const pending: Promise[] = []; + walk_coresdk_ActivityTaskCompletion(root, env, context, pending); + return pending; +} + +export function walkNexusTask( + root: coresdk.nexus.INexusTask, + env: WalkEnv, + context: Ctx +): Promise[] { + const pending: Promise[] = []; + walk_coresdk_nexus_NexusTask(root, env, context, pending); + return pending; +} + +export function walkNexusTaskCompletion( + root: coresdk.nexus.INexusTaskCompletion, + env: WalkEnv, + context: Ctx +): Promise[] { + const pending: Promise[] = []; + walk_coresdk_nexus_NexusTaskCompletion(root, env, context, pending); + return pending; +} + export function walkWorkflowActivation( root: coresdk.workflow_activation.IWorkflowActivation, env: WalkEnv, @@ -32,6 +82,29 @@ export function walkWorkflowActivationCompletion( return pending; } +function walk_coresdk_activity_result_ActivityExecutionResult( + o: coresdk.activity_result.IActivityExecutionResult, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext + ? env.deriveContext(o, 'coresdk.activity_result.ActivityExecutionResult', context) + : context; + { + const c = o.completed; + if (c != null) walk_coresdk_activity_result_Success(c, env, ctx, pending); + } + { + const c = o.failed; + if (c != null) walk_coresdk_activity_result_Failure(c, env, ctx, pending); + } + { + const c = o.cancelled; + if (c != null) walk_coresdk_activity_result_Cancellation(c, env, ctx, pending); + } +} + function walk_coresdk_activity_result_ActivityResolution( o: coresdk.activity_result.IActivityResolution, env: WalkEnv, @@ -97,6 +170,87 @@ function walk_coresdk_activity_result_Success( } } +function walk_coresdk_activity_task_ActivityTask( + o: coresdk.activity_task.IActivityTask, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'coresdk.activity_task.ActivityTask', context) : context; + { + const c = o.start; + if (c != null) walk_coresdk_activity_task_Start(c, env, ctx, pending); + } +} + +function walk_coresdk_activity_task_Start( + o: coresdk.activity_task.IStart, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'coresdk.activity_task.Start', context) : context; + { + const m = o.headerFields; + if (m) + for (const [k, v] of Object.entries(m)) + pending.push( + env.transformPayload(v, ctx).then((r) => { + m[k] = r; + }) + ); + } + { + const a = o.input; + if (a && a.length) + pending.push( + env.transformPayloads(a, ctx).then((r) => { + o.input = r; + }) + ); + } + { + const a = o.heartbeatDetails; + if (a && a.length) + pending.push( + env.transformPayloads(a, ctx).then((r) => { + o.heartbeatDetails = r; + }) + ); + } +} + +function walk_coresdk_ActivityHeartbeat( + o: coresdk.IActivityHeartbeat, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'coresdk.ActivityHeartbeat', context) : context; + { + const a = o.details; + if (a && a.length) + pending.push( + env.transformPayloads(a, ctx).then((r) => { + o.details = r; + }) + ); + } +} + +function walk_coresdk_ActivityTaskCompletion( + o: coresdk.IActivityTaskCompletion, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'coresdk.ActivityTaskCompletion', context) : context; + { + const c = o.result; + if (c != null) walk_coresdk_activity_result_ActivityExecutionResult(c, env, ctx, pending); + } +} + function walk_coresdk_child_workflow_Cancellation( o: coresdk.child_workflow.ICancellation, env: WalkEnv, @@ -192,6 +346,36 @@ function walk_coresdk_nexus_NexusOperationResult( } } +function walk_coresdk_nexus_NexusTask( + o: coresdk.nexus.INexusTask, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'coresdk.nexus.NexusTask', context) : context; + { + const c = o.task; + if (c != null) walk_temporal_api_workflowservice_v1_PollNexusTaskQueueResponse(c, env, ctx, pending); + } +} + +function walk_coresdk_nexus_NexusTaskCompletion( + o: coresdk.nexus.INexusTaskCompletion, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'coresdk.nexus.NexusTaskCompletion', context) : context; + { + const c = o.completed; + if (c != null) walk_temporal_api_nexus_v1_Response(c, env, ctx, pending); + } + { + const c = o.failure; + if (c != null) walk_temporal_api_failure_v1_Failure(c, env, ctx, pending); + } +} + function walk_coresdk_workflow_activation_DoUpdate( o: coresdk.workflow_activation.IDoUpdate, env: WalkEnv, @@ -1114,6 +1298,91 @@ function walk_temporal_api_failure_v1_TimeoutFailureInfo( } } +function walk_temporal_api_nexus_v1_Request( + o: temporal.api.nexus.v1.IRequest, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'temporal.api.nexus.v1.Request', context) : context; + { + const c = o.startOperation; + if (c != null) walk_temporal_api_nexus_v1_StartOperationRequest(c, env, ctx, pending); + } +} + +function walk_temporal_api_nexus_v1_Response( + o: temporal.api.nexus.v1.IResponse, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext ? env.deriveContext(o, 'temporal.api.nexus.v1.Response', context) : context; + { + const c = o.startOperation; + if (c != null) walk_temporal_api_nexus_v1_StartOperationResponse(c, env, ctx, pending); + } +} + +function walk_temporal_api_nexus_v1_StartOperationRequest( + o: temporal.api.nexus.v1.IStartOperationRequest, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext + ? env.deriveContext(o, 'temporal.api.nexus.v1.StartOperationRequest', context) + : context; + { + const p = o.payload; + if (p != null) + pending.push( + env.transformPayload(p, ctx).then((r) => { + o.payload = r; + }) + ); + } +} + +function walk_temporal_api_nexus_v1_StartOperationResponse( + o: temporal.api.nexus.v1.IStartOperationResponse, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext + ? env.deriveContext(o, 'temporal.api.nexus.v1.StartOperationResponse', context) + : context; + { + const c = o.syncSuccess; + if (c != null) walk_temporal_api_nexus_v1_StartOperationResponse_Sync(c, env, ctx, pending); + } + { + const c = o.failure; + if (c != null) walk_temporal_api_failure_v1_Failure(c, env, ctx, pending); + } +} + +function walk_temporal_api_nexus_v1_StartOperationResponse_Sync( + o: temporal.api.nexus.v1.StartOperationResponse.ISync, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext + ? env.deriveContext(o, 'temporal.api.nexus.v1.StartOperationResponse.Sync', context) + : context; + { + const p = o.payload; + if (p != null) + pending.push( + env.transformPayload(p, ctx).then((r) => { + o.payload = r; + }) + ); + } +} + function walk_temporal_api_sdk_v1_UserMetadata( o: temporal.api.sdk.v1.IUserMetadata, env: WalkEnv, @@ -1140,3 +1409,18 @@ function walk_temporal_api_sdk_v1_UserMetadata( ); } } + +function walk_temporal_api_workflowservice_v1_PollNexusTaskQueueResponse( + o: temporal.api.workflowservice.v1.IPollNexusTaskQueueResponse, + env: WalkEnv, + context: Ctx, + pending: Promise[] +): void { + const ctx = env.deriveContext + ? env.deriveContext(o, 'temporal.api.workflowservice.v1.PollNexusTaskQueueResponse', context) + : context; + { + const c = o.request; + if (c != null) walk_temporal_api_nexus_v1_Request(c, env, ctx, pending); + } +}