From 89405a4610b7c7fc05f7aab28d50355b06e61263 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Mon, 14 Sep 2026 11:37:20 +0200 Subject: [PATCH] feat(temporal): support per-node-type taskQueue routing (#141) --- apps/execution-worker/.env.example | 4 + apps/execution-worker/README.md | 15 ++++ apps/execution-worker/package.json | 1 + .../engines/temporal/specialized-worker.ts | 44 +++++++++ apps/execution-worker/src/env.ts | 3 + deploy/ai-studio/README.md | 41 ++++++--- deploy/ai-studio/docker-compose.yml | 24 +++++ packages/temporal/README.md | 17 ++++ .../src/workflow/activity-profiles.ts | 4 + .../workflow/node-activity-options.test.ts | 34 +++++++ .../src/workflow/node-activity-options.ts | 1 + .../src/workflow/profile-validation.test.ts | 58 +++++++++++- .../src/workflow/profile-validation.ts | 21 ++++- .../fixtures/workflows-with-task-queue.ts | 12 +++ .../temporal/test/task-queue-routing.test.ts | 89 +++++++++++++++++++ 15 files changed, 351 insertions(+), 17 deletions(-) create mode 100644 apps/execution-worker/src/engines/temporal/specialized-worker.ts create mode 100644 packages/temporal/test/fixtures/workflows-with-task-queue.ts create mode 100644 packages/temporal/test/task-queue-routing.test.ts diff --git a/apps/execution-worker/.env.example b/apps/execution-worker/.env.example index 94e2ecd4c..7c17aee7f 100644 --- a/apps/execution-worker/.env.example +++ b/apps/execution-worker/.env.example @@ -9,3 +9,7 @@ AI_MODEL=mistralai/mistral-small-3.2-24b-instruct # free key at https://tavily.com (free tier ~1000 searches/month). Leave empty # to disable: agents with web search toggled on still run, just without the tool. TAVILY_API_KEY= + +# Queue polled only by the specialized worker (start:specialized). A node type is +# routed here via its activity profile's `taskQueue`; unset means the default queue. +SPECIALIZED_TASK_QUEUE=workflow-execution-specialized diff --git a/apps/execution-worker/README.md b/apps/execution-worker/README.md index a9aa5d107..1eaed9c87 100644 --- a/apps/execution-worker/README.md +++ b/apps/execution-worker/README.md @@ -44,6 +44,7 @@ src/ └── engines/ └── temporal/ ├── worker.ts # Worker bootstrap: executors + store, handed to WorkflowBuilderPlugin + ├── specialized-worker.ts # Activity-only worker for a taskQueue-routed subset of node types └── workflows.ts # One-line re-export of runWorkflow for Temporal's bundler ``` @@ -51,6 +52,20 @@ The workflow itself, the activity contract and the event emitter live in [`@workflowbuilder/temporal`](../../packages/temporal/README.md). This app only supplies what is its own: one executor per node type and the database as the store port. +## Per-node-type task queue routing + +A node type's activity profile can carry `taskQueue`, which pins it to a queue other +than the default (`plugin.taskQueue`). `worker.ts` keeps polling the default queue for +everything else; `specialized-worker.ts` is a second, activity-only entrypoint (no +`workflowsPath` — Temporal supports activity-only workers) that polls +`SPECIALIZED_TASK_QUEUE` and registers only the node type(s) routed there. Run it with +`pnpm --filter execution-worker start:specialized`, or as the `worker-specialized` +compose service (see `deploy/ai-studio/README.md`). + +This is a deployment-only change: `runGraph` and the graph model never see a taskQueue, +they only affect which worker process a node's `executeNode` activity is scheduled on. +A profile with no `taskQueue` behaves exactly as before. + ## Temporal specifics - **Task queue:** `workflow-execution`, read from `plugin.taskQueue` so the backend and the worker cannot drift apart. Both default to the same constant in the package. diff --git a/apps/execution-worker/package.json b/apps/execution-worker/package.json index b0ee15a9b..18d78d311 100644 --- a/apps/execution-worker/package.json +++ b/apps/execution-worker/package.json @@ -7,6 +7,7 @@ "dev": "tsx watch --env-file=.env ./src/engines/temporal/worker.ts", "start": "tsx --env-file=.env ./src/engines/temporal/worker.ts", "start:prod": "tsx ./src/engines/temporal/worker.ts", + "start:specialized": "tsx ./src/engines/temporal/specialized-worker.ts", "typecheck": "tsc --noEmit", "lint": "eslint", "lint:fix": "eslint --fix", diff --git a/apps/execution-worker/src/engines/temporal/specialized-worker.ts b/apps/execution-worker/src/engines/temporal/specialized-worker.ts new file mode 100644 index 000000000..a834db457 --- /dev/null +++ b/apps/execution-worker/src/engines/temporal/specialized-worker.ts @@ -0,0 +1,44 @@ +// Activity-only worker: polls SPECIALIZED_TASK_QUEUE and registers just the node +// type(s) routed there via a profile's `taskQueue`. No `workflowsPath` — Temporal +// supports activity-only workers, so this process never needs the workflow bundle. +// Runs from its own Docker image so its tools stay off the general worker's image. +import { NativeConnection, Worker } from '@temporalio/worker'; +import { WorkflowBuilderPlugin } from '@workflowbuilder/temporal'; +import 'dotenv/config'; + +import { executeAiAgent } from '../../activities/ai-agent'; +import { database } from '../../database'; +import type { AiAgentNode } from '../../domain/ai-studio-nodes'; +import { env } from '../../env'; +import { logger } from '../../logger'; +import { withPayloadSizeWarning } from '../../store-payload-warning'; + +const { createOpenRouter } = await import('@openrouter/ai-sdk-provider'); + +const openrouter = createOpenRouter({ apiKey: env.OPENROUTER_API_KEY }); +const model = openrouter.chat(env.AI_MODEL); + +const aiAgentLogger = logger.child({ component: 'ai-agent', worker: 'specialized' }); + +// Illustrative: the AI agent is the reference example of a node type heavy enough to +// want its own image (a coding-agent CLI, GPU tooling, ...). Swap in whichever +// executor(s) a real deployment routes to this queue via nodeActivityProfiles. +const plugin = new WorkflowBuilderPlugin({ + executors: { + 'ai-studio/ai-agent': (node, context) => + executeAiAgent(node, context, { model, logger: aiAgentLogger, tavilyApiKey: env.TAVILY_API_KEY }), + }, + store: withPayloadSizeWarning(database, logger), + taskQueue: env.SPECIALIZED_TASK_QUEUE, +}); + +const connection = await NativeConnection.connect({ address: env.TEMPORAL_ADDRESS }); + +const worker = await Worker.create({ + connection, + taskQueue: plugin.taskQueue, + plugins: [plugin], +}); + +logger.info('specialized execution worker started', { taskQueue: plugin.taskQueue }); +await worker.run(); diff --git a/apps/execution-worker/src/env.ts b/apps/execution-worker/src/env.ts index 5bbd6a61a..970f422b7 100644 --- a/apps/execution-worker/src/env.ts +++ b/apps/execution-worker/src/env.ts @@ -21,4 +21,7 @@ export const env = { AI_MODEL: envOr('AI_MODEL', 'mistralai/mistral-small-3.2-24b-instruct'), // Optional. Enables the AI Agent's web-search tool; agents run without it when unset. TAVILY_API_KEY: process.env['TAVILY_API_KEY'], + // Polled only by the specialized worker (see engines/temporal/specialized-worker.ts). + // A distinct queue name, not the plugin's own, so routing a node here is opt-in per type. + SPECIALIZED_TASK_QUEUE: envOr('SPECIALIZED_TASK_QUEUE', 'workflow-execution-specialized'), }; diff --git a/deploy/ai-studio/README.md b/deploy/ai-studio/README.md index 530155ac3..2aa5ea90a 100644 --- a/deploy/ai-studio/README.md +++ b/deploy/ai-studio/README.md @@ -5,21 +5,36 @@ any Docker host — an Azure VM, AWS, on-prem — with no cloud-specific glue. ## What runs -| Service | Image | Role | Exposed | -| ------------- | ------------------------------ | ----------------------------------------------- | ------------------------ | -| `web` | `ai-studio-web` (nginx) | Serves the SPA, proxies `/api` to the backend | `${WEB_PORT}` (only one) | -| `backend` | `ai-studio-runtime` | Hono REST + SSE event stream | internal | -| `worker` | `ai-studio-runtime` | Temporal worker, makes the OpenRouter LLM calls | internal | -| `temporal` | `temporalio/auto-setup` pinned | Workflow engine | internal | -| `app-db` | `postgres:16` | Workflow snapshots + execution events | internal | -| `temporal-db` | `postgres:16` | Temporal's own state store | internal | -| `temporal-ui` | `temporalio/ui` pinned | Debug only (`--profile debug`) | `127.0.0.1:8233` | +| Service | Image | Role | Exposed | +| -------------------- | ------------------------------ | ---------------------------------------------------------------------- | ------------------------ | +| `web` | `ai-studio-web` (nginx) | Serves the SPA, proxies `/api` to the backend | `${WEB_PORT}` (only one) | +| `backend` | `ai-studio-runtime` | Hono REST + SSE event stream | internal | +| `worker` | `ai-studio-runtime` | Temporal worker, makes the OpenRouter LLM calls | internal | +| `worker-specialized` | `ai-studio-runtime` | Activity-only worker for node types routed to `SPECIALIZED_TASK_QUEUE` | internal | +| `temporal` | `temporalio/auto-setup` pinned | Workflow engine | internal | +| `app-db` | `postgres:16` | Workflow snapshots + execution events | internal | +| `temporal-db` | `postgres:16` | Temporal's own state store | internal | +| `temporal-ui` | `temporalio/ui` pinned | Debug only (`--profile debug`) | `127.0.0.1:8233` | Both images build from one Dockerfile (`deploy/ai-studio/Dockerfile`) with the -repo root as context. Backend and worker share a single image and differ only -in the compose `command`. Database migrations are applied by the backend at -boot (drizzle-orm's programmatic migrator) — there is no separate migration -service or step. +repo root as context. Backend, worker and worker-specialized share a single +image and differ only in the compose `command`. Database migrations are +applied by the backend at boot (drizzle-orm's programmatic migrator) — there +is no separate migration service or step. + +`worker-specialized` polls its own task queue and registers only the node +type(s) meant to run on it (see `apps/execution-worker/README.md` § "Per-node-type +task queue routing"). It only executes anything once a node's activity profile +sets `taskQueue: 'workflow-execution-specialized'` (or your own +`SPECIALIZED_TASK_QUEUE`); until then it idles, polling a queue nothing is scheduled to. + +### Running the specialized worker standalone + +```bash +pnpm --filter execution-worker start:specialized # needs infra up + SPECIALIZED_TASK_QUEUE set +``` + +Or in compose alone: `docker compose up -d worker-specialized`. ## Quick start diff --git a/deploy/ai-studio/docker-compose.yml b/deploy/ai-studio/docker-compose.yml index 5eed67bf7..eb075e96f 100644 --- a/deploy/ai-studio/docker-compose.yml +++ b/deploy/ai-studio/docker-compose.yml @@ -126,6 +126,30 @@ services: condition: service_started restart: unless-stopped + # dedicated pool for node types routed via a profile's `taskQueue`, so their + # dependencies (heavier tools, a different runtime) never reach the shared image. + # Same runtime image today — give it its own Dockerfile stage only once a routed + # node type actually needs extra OS-level tooling the shared image doesn't have. + worker-specialized: + image: ai-studio-runtime + build: *runtime-build + command: ['pnpm', '--filter', 'execution-worker', 'start:specialized'] + environment: + DATABASE_URL: postgresql://wb:${APP_DB_PASSWORD:-wb}@app-db:5432/workflow_builder + TEMPORAL_ADDRESS: temporal:7233 + OPENROUTER_API_KEY: ${OPENROUTER_API_KEY:?set OPENROUTER_API_KEY in deploy/ai-studio/.env} + AI_MODEL: ${AI_MODEL:-mistralai/mistral-small-3.2-24b-instruct} + TAVILY_API_KEY: ${TAVILY_API_KEY:-} + SPECIALIZED_TASK_QUEUE: ${SPECIALIZED_TASK_QUEUE:-workflow-execution-specialized} + depends_on: + app-db: + condition: service_healthy + backend: + condition: service_healthy + temporal: + condition: service_started + restart: unless-stopped + web: image: ai-studio-web build: diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 0d94a4bad..1be5f58bf 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -95,6 +95,23 @@ Keep the export named `runWorkflow`: that is the name the client starts, and a t Entries are whole profiles rather than partials on purpose. A partial would let you set a timeout and silently drop the retry cap, and what Temporal falls back to is unlimited retries with backoff, which on a permanently failing model call is an unbounded bill. A node type with no entry resolves to `DEFAULT_NODE_ACTIVITY_PROFILE` and nothing else. +A profile can also carry `taskQueue`, routing that node type's activities to a worker other than the one running the workflow — a small pool with its own Docker image and tools, while every other node type keeps running on the default queue. Nothing else about the workflow changes: `runGraph` and the graph model never see it, it only decides which worker process picks up the `executeNode` command. Without a `taskQueue`, a profile behaves exactly as it did before this field existed. + +```ts +export const nodeActivityProfiles: NodeActivityProfiles = { + // Needs a coding-agent CLI installed — its own image, its own worker. + 'my-product/coding-agent': { + startToCloseTimeout: '30m', + retry: { maximumAttempts: 2 }, + taskQueue: 'coding-agent-pool', + }, +}; +``` + +A worker for that queue is a second, activity-only `Worker.create` call (no `workflowsPath` — Temporal supports activity-only workers) registering just that node type's executor. See `apps/execution-worker/README.md` § "Per-node-type task queue routing" for a full example. + +A `taskQueue` routed to a queue nobody polls fails silently at the workflow level: the activity is scheduled and simply never picked up. Checking that is only possible worker-side, against whichever queues a deployment knows it runs — `findProfilesWithUnpolledTaskQueue(profiles, polledTaskQueues)` from `/workflow` is the same shape as `findProfilesWithoutExecutor`, for the same reason: the sandbox has no registry to check it against itself. + A `startToCloseTimeout` is a number followed by `ms`, `s`, `m`, `h` or `d`. Decimals are fine (`'1.5h'`). It has to fit a protobuf `Duration`, so anything under one nanosecond or over `'3652500d'` is out. Zero, negative values and exponent notation are rejected even though TypeScript's template literal type admits them: `'0s'` type-checks, and Temporal treats a zero timeout as unset and refuses to schedule the activity. This grammar is narrower than Temporal's own, which parses durations with the `ms` package and also takes `'30 minutes'` or `'1 week'`. One documented form is deliberate. If you think in the wider grammar, convert before the value reaches this map. diff --git a/packages/temporal/src/workflow/activity-profiles.ts b/packages/temporal/src/workflow/activity-profiles.ts index e18b6a301..189f72aad 100644 --- a/packages/temporal/src/workflow/activity-profiles.ts +++ b/packages/temporal/src/workflow/activity-profiles.ts @@ -12,6 +12,9 @@ export type DurationString = `${number}${'ms' | 's' | 'm' | 'h' | 'd'}`; export type ActivityProfile = { startToCloseTimeout: DurationString; retry: { maximumAttempts: number }; + // Routes this node type's activities to a non-default queue, e.g. a specialized + // worker with its own image/tools. Absent means the plugin's own task queue. + taskQueue?: string; }; // Only for the two frozen singletons below. Annotating them `ActivityProfile` would @@ -20,6 +23,7 @@ export type ActivityProfile = { type ReadonlyActivityProfile = { readonly startToCloseTimeout: DurationString; readonly retry: { readonly maximumAttempts: number }; + readonly taskQueue?: string; }; // Node activities may call LLMs (minutes) — generous timeout, fewer retries to limit diff --git a/packages/temporal/src/workflow/node-activity-options.test.ts b/packages/temporal/src/workflow/node-activity-options.test.ts index 482b60683..20d1b0678 100644 --- a/packages/temporal/src/workflow/node-activity-options.test.ts +++ b/packages/temporal/src/workflow/node-activity-options.test.ts @@ -129,6 +129,40 @@ describe('resolveNodeActivityOptions', () => { }); }); + describe('taskQueue', () => { + it('forwards taskQueue when the profile carries one', () => { + const profiles: NodeActivityProfiles = { + 'test/step': { startToCloseTimeout: '2m', retry: { maximumAttempts: 4 }, taskQueue: 'specialized' }, + }; + + expect(resolveNodeActivityOptions(node(), profiles)).toEqual({ + startToCloseTimeout: '2m', + retry: { maximumAttempts: 4 }, + taskQueue: 'specialized', + }); + }); + + it('omits the key entirely when the profile has none, not taskQueue: undefined', () => { + const resolved = resolveNodeActivityOptions(node(), {}); + + expect('taskQueue' in resolved).toBe(false); + }); + + it('is pure routing metadata: identical graph-relevant fields whether taskQueue is set or not', () => { + // The workflow sandbox must not branch on taskQueue — it only reaches proxyActivities. + const base = resolveNodeActivityOptions(node(), { + 'test/step': { startToCloseTimeout: '2m', retry: { maximumAttempts: 4 } }, + }); + const routed = resolveNodeActivityOptions(node(), { + 'test/step': { startToCloseTimeout: '2m', retry: { maximumAttempts: 4 }, taskQueue: 'specialized' }, + }); + + expect(routed.startToCloseTimeout).toBe(base.startToCloseTimeout); + expect(routed.retry).toEqual(base.retry); + expect(routed.summary).toBe(base.summary); + }); + }); + describe('summary', () => { it('carries the node label so Event History reads like the diagram', () => { const resolved = resolveNodeActivityOptions(node({ label: 'Fetch order' }), {}); diff --git a/packages/temporal/src/workflow/node-activity-options.ts b/packages/temporal/src/workflow/node-activity-options.ts index b430ac0b8..5cb12223e 100644 --- a/packages/temporal/src/workflow/node-activity-options.ts +++ b/packages/temporal/src/workflow/node-activity-options.ts @@ -41,6 +41,7 @@ export function resolveFromValidatedProfiles(node: BaseNode, profiles: NodeActiv const options: NodeActivityOptions = { startToCloseTimeout: profile.startToCloseTimeout, retry: { maximumAttempts: profile.retry.maximumAttempts }, + ...(profile.taskQueue === undefined ? {} : { taskQueue: profile.taskQueue }), }; // Not trusted from the caller: any consumer can build the workflow input. Temporal diff --git a/packages/temporal/src/workflow/profile-validation.test.ts b/packages/temporal/src/workflow/profile-validation.test.ts index 28e2fdd33..f08dd65f2 100644 --- a/packages/temporal/src/workflow/profile-validation.test.ts +++ b/packages/temporal/src/workflow/profile-validation.test.ts @@ -5,6 +5,7 @@ import type { BaseNode } from './core-contract'; import { resolveNodeActivityOptions } from './node-activity-options'; import { assertNodeActivityProfiles, + findProfilesWithUnpolledTaskQueue, findProfilesWithoutExecutor, freezeNodeActivityProfiles, } from './profile-validation'; @@ -129,13 +130,43 @@ describe('assertNodeActivityProfiles', () => { 'test/step': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 }, - taskQueue: 'x', + scheduleToCloseTimeout: '1h', heartbeatTimeout: '1m', }, }; expect(() => assertNodeActivityProfiles(extra as unknown as NodeActivityProfiles)).toThrow( - /has unknown keys "taskQueue", "heartbeatTimeout"/, + /has unknown keys "scheduleToCloseTimeout", "heartbeatTimeout"/, + ); + }); + }); + + describe('taskQueue', () => { + it('accepts a profile with a non-empty taskQueue', () => { + const withQueue = { + 'test/step': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 }, taskQueue: 'specialized' }, + }; + + expect(() => assertNodeActivityProfiles(withQueue as unknown as NodeActivityProfiles)).not.toThrow(); + }); + + it('accepts a profile with no taskQueue, unchanged from before', () => { + expect(() => assertNodeActivityProfiles(profiles('10m'))).not.toThrow(); + }); + + it('rejects an empty string, naming the path', () => { + const empty = { 'test/step': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 }, taskQueue: '' } }; + + expect(() => assertNodeActivityProfiles(empty as unknown as NodeActivityProfiles)).toThrow( + /nodeActivityProfiles\["test\/step"\]\.taskQueue must be a non-empty string/, + ); + }); + + it('rejects a non-string value', () => { + const wrongType = { 'test/step': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 }, taskQueue: 7 } }; + + expect(() => assertNodeActivityProfiles(wrongType as unknown as NodeActivityProfiles)).toThrow( + /taskQueue must be a non-empty string/, ); }); }); @@ -205,3 +236,26 @@ describe('findProfilesWithoutExecutor', () => { expect(findProfilesWithoutExecutor(inherited, executors)).toEqual(['constructor']); }); }); + +describe('findProfilesWithUnpolledTaskQueue', () => { + it('names a node type routed to a queue nobody polls', () => { + const map = { + 'test/routed': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 }, taskQueue: 'specialized' }, + 'test/default': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 } }, + } as unknown as NodeActivityProfiles; + + expect(findProfilesWithUnpolledTaskQueue(map, new Set(['workflow-execution']))).toEqual(['test/routed']); + }); + + it('is silent once the queue is in the polled set', () => { + const map = { + 'test/routed': { startToCloseTimeout: '10m', retry: { maximumAttempts: 2 }, taskQueue: 'specialized' }, + } as unknown as NodeActivityProfiles; + + expect(findProfilesWithUnpolledTaskQueue(map, new Set(['specialized']))).toEqual([]); + }); + + it('ignores a profile with no taskQueue regardless of the polled set', () => { + expect(findProfilesWithUnpolledTaskQueue(profiles('10m'), new Set())).toEqual([]); + }); +}); diff --git a/packages/temporal/src/workflow/profile-validation.ts b/packages/temporal/src/workflow/profile-validation.ts index e8112dce7..6b231b67d 100644 --- a/packages/temporal/src/workflow/profile-validation.ts +++ b/packages/temporal/src/workflow/profile-validation.ts @@ -42,11 +42,15 @@ function assertActivityProfile(nodeType: string, profile: ActivityProfile | unde ); } + if (profile.taskQueue !== undefined && (typeof profile.taskQueue !== 'string' || profile.taskQueue === '')) { + throw new TypeError(`${path}.taskQueue must be a non-empty string, got ${JSON.stringify(profile.taskQueue)}.`); + } + assertNoUnknownKeys(path, profile, PROFILE_KEYS); assertNoUnknownKeys(`${path}.retry`, profile.retry, RETRY_KEYS); } -const PROFILE_KEYS: ReadonlySet = new Set(['startToCloseTimeout', 'retry']); +const PROFILE_KEYS: ReadonlySet = new Set(['startToCloseTimeout', 'retry', 'taskQueue']); const RETRY_KEYS: ReadonlySet = new Set(['maximumAttempts']); function assertNoUnknownKeys(path: string, value: object, allowed: ReadonlySet): void { @@ -54,7 +58,7 @@ function assertNoUnknownKeys(path: string, value: object, allowed: ReadonlySet JSON.stringify(key)).join(', ')}. A profile carries startToCloseTimeout and retry.maximumAttempts, nothing else.`, + `${path} has unknown ${unknown.length === 1 ? 'key' : 'keys'} ${unknown.map((key) => JSON.stringify(key)).join(', ')}. A profile carries startToCloseTimeout, retry.maximumAttempts and an optional taskQueue, nothing else.`, ); } @@ -83,3 +87,16 @@ export function freezeNodeActivityProfiles(profiles: NodeActivityProfiles): Node export function findProfilesWithoutExecutor(profiles: NodeActivityProfiles, executors: object): string[] { return Object.keys(profiles).filter((nodeType) => !Object.hasOwn(executors, nodeType)); } + +// Same shape as findProfilesWithoutExecutor: only checkable worker-side, since a +// profile's taskQueue is routing metadata the sandbox never sees. `polledTaskQueues` +// is whatever queues this call knows are actually polled (e.g. this worker's own plus +// any others it is told about) — a node routed elsewhere silently never runs. +export function findProfilesWithUnpolledTaskQueue( + profiles: NodeActivityProfiles, + polledTaskQueues: ReadonlySet, +): string[] { + return Object.entries(profiles) + .filter(([, profile]) => profile.taskQueue !== undefined && !polledTaskQueues.has(profile.taskQueue)) + .map(([nodeType]) => nodeType); +} diff --git a/packages/temporal/test/fixtures/workflows-with-task-queue.ts b/packages/temporal/test/fixtures/workflows-with-task-queue.ts new file mode 100644 index 000000000..b6930e5b0 --- /dev/null +++ b/packages/temporal/test/fixtures/workflows-with-task-queue.ts @@ -0,0 +1,12 @@ +// A second workflows module, built with createRunWorkflow, so a test can prove a +// profile's taskQueue reaches the ScheduleActivityTask command. The zero-config +// export in ./workflows.ts cannot exercise this: it never calls createRunWorkflow. +import { createRunWorkflow } from '../../src/workflow/index'; + +export const SPECIALIZED_TASK_QUEUE = 'test-specialized-queue'; + +export const runWorkflow = createRunWorkflow({ + nodeActivityProfiles: { + 'test/step': { startToCloseTimeout: '10s', retry: { maximumAttempts: 1 }, taskQueue: SPECIALIZED_TASK_QUEUE }, + }, +}); diff --git a/packages/temporal/test/task-queue-routing.test.ts b/packages/temporal/test/task-queue-routing.test.ts new file mode 100644 index 000000000..c5beab83a --- /dev/null +++ b/packages/temporal/test/task-queue-routing.test.ts @@ -0,0 +1,89 @@ +// Proves item 6 of the taskQueue-routing issue: a profile's taskQueue reaches the +// real ScheduleActivityTask command, not just the resolved options object. +import { type History } from '@temporalio/common/lib/proto-utils'; +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker, bundleWorkflowCode } from '@temporalio/worker'; +import { fileURLToPath } from 'node:url'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import { WorkflowBuilderPlugin, type WorkflowExecutionInput, executionWorkflowId } from '../src/index'; +import { REPLAY_TEST_GRAPH, createRecordingStore, replayTestExecutors } from './fixtures/graph'; +import { SPECIALIZED_TASK_QUEUE } from './fixtures/workflows-with-task-queue'; + +const EXECUTION_ID = 'task-queue-routing-execution'; +const TASK_QUEUE = 'task-queue-routing-test'; + +describe('taskQueue routing', () => { + let env: TestWorkflowEnvironment; + let history: History; + + beforeAll(async () => { + const [workflowBundle, testEnv] = await Promise.all([ + bundleWorkflowCode({ + workflowsPath: fileURLToPath(new URL('fixtures/workflows-with-task-queue.ts', import.meta.url)), + }), + TestWorkflowEnvironment.createLocal(), + ]); + env = testEnv; + + const plugin = new WorkflowBuilderPlugin({ + store: createRecordingStore(), + executors: replayTestExecutors, + taskQueue: TASK_QUEUE, + }); + + // Polls only the default queue: the routed node's activity is deliberately never + // executed, so the workflow stays open while its ScheduleActivityTask is inspected. + const worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: plugin.taskQueue, + workflowBundle, + plugins: [plugin], + }); + + const input: WorkflowExecutionInput<(typeof REPLAY_TEST_GRAPH)['nodes'][number]> = { + workflowId: REPLAY_TEST_GRAPH.workflowId, + executionId: EXECUTION_ID, + definition: REPLAY_TEST_GRAPH, + triggerPayload: {}, + variables: {}, + global: {}, + }; + + const workflowId = executionWorkflowId(EXECUTION_ID); + + await worker.runUntil(async () => { + const handle = await env.client.workflow.start('runWorkflow', { + taskQueue: plugin.taskQueue, + workflowId, + args: [input], + }); + + // The workflow awaits the execution_started DB activity (on the default queue) + // before scheduling the routed node, so history is polled until that command lands. + for (let attempt = 0; attempt < 50; attempt += 1) { + history = await handle.fetchHistory(); + const hasExecuteNode = (history.events ?? []).some( + (event) => event.activityTaskScheduledEventAttributes?.activityType?.name === 'executeNode', + ); + if (hasExecuteNode) break; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + + await handle.terminate('test cleanup'); + }); + }, 60_000); + + afterAll(async () => { + await env?.teardown(); + }); + + it('schedules the routed node on its profile-declared taskQueue, not the workflow default', () => { + const scheduled = (history.events ?? []).find( + (event) => event.activityTaskScheduledEventAttributes?.activityType?.name === 'executeNode', + ); + + expect(scheduled?.activityTaskScheduledEventAttributes?.taskQueue?.name).toBe(SPECIALIZED_TASK_QUEUE); + }); +});