Temporal primitives · session note

Temporal queues & options — per-activity routing, per-queue tuning, worker-runtime profiles

The companion to id-and-naming.html. Where the ID work was about identity, this work is about knobs: which task queue an activity dispatches to, what timeouts and retry policy apply, how the worker process itself is tuned. Three independent config layers, each addressing a real operational pain.

1. The three things this layer lets you do

CAPABILITY #1

Route by activity and by handle

Send each activity (and optionally each handle within an activity) to a dedicated task queue. Isolate GPU-bound image-gen runners from cheap LLM workers, route a specific OCR backend to a dedicated queue, scale OpenAI vs Anthropic pools independently.

CAPABILITY #2

Tune dispatch per queue and per handle

Set start_to_close_timeout, retry policy, heartbeat, and rate caps per task queue. Override for a single handle when one model variant on a backend needs different tuning. The workflow's own deadline is no longer spent by a single in-flight activity.

CAPABILITY #3

Specialize worker processes

Named worker-runtime profiles bundle the Worker(...) tuning (slots, pollers, heartbeat throttles, graceful shutdown). One worker process picks one profile via pipelex worker --profile <name>, plus a --scope selecting which activities it registers.

2. The shape of the solution — three orthogonal layers

Layer A
where it goes
activity_queues

Layer B
how it's dispatched
queue_options
handle_options

Layer C
how the worker behaves
worker_runtime_profiles

Layer A — routing. Per activity, per handle, which task queue do we dispatch to? Submitter-side.
Layer B — options. For that resolved queue (and optionally that specific handle), what timeouts and retry policy apply? Submitter-side.
Layer C — worker. Inside one worker process, how do we tune slots/pollers/heartbeats? Worker-side, selected via --profile.

3. Layer A — per-activity, per-handle routing

The activity_queues table on WorkerConfig is the entry point. Each entry declares a default queue for one activity and an optional by_handle map keyed by runtime handle (LLM model handle, image-gen handle, extract handle).

# pipelex.toml
[temporal.worker_config]
default_task_queue = "temporal_task_queue"
default_activity_start_to_close_timeout = "0:10:00"

# Route LLM text by model family. Anthropic runs on a dedicated pool;
# everything else falls back to the activity default.
[temporal.worker_config.activity_queues.act_llm_gen_text]
default = "llm-default-queue"
by_handle = {
  "claude-4.6-sonnet" = "anthropic-queue",
  "claude-4.6-opus"   = "anthropic-queue",
  "gpt-5.1"           = "openai-queue",
}

# Image generation on a GPU pool.
[temporal.worker_config.activity_queues.act_img_gen_images]
default = "img-gen-gpu-queue"

# Extract: route Azure Doc Intel separately from Mistral OCR.
[temporal.worker_config.activity_queues.act_extract_gen_extract_pages]
default = "extract-default-queue"
by_handle = { "azure-doc-intel" = "extract-azure-queue" }

The resolver — three steps, last-wins

WorkerConfig.resolve_queue(activity_name, routing_key) -> str | None

  1. activity_queues[activity_name].by_handle[routing_key]   # most specific
  2. activity_queues[activity_name].default                  # activity-level default
  3. self.default_task_queue                                 # worker-wide default

Hybrid fallback semantic. When activity_queues is fully empty (default config, no routing configured), resolve_queue returns None — and the dispatch path then omits the task_queue kwarg so Temporal routes the activity to the workflow's own queue. This preserves the pre-v1 single-queue default and the with_conditional_worker test pattern. As soon as any activity_queues entry exists, the operator has opted into routing topology and unmapped activities fall back explicitly to default_task_queue.

4. Layer B — per-queue and per-handle options

queue_options declares the submitter options (timeouts, retry policy, queue-level rate cap) for one task queue. handle_options (inside an activity_queues entry) declares the rare per-handle override on top of the queue's options.

# pipelex.toml — queue-level submitter options
[temporal.queue_options.temporal_task_queue]
start_to_close_timeout = "0:10:00"
heartbeat_timeout      = "0:01:00"
max_task_queue_activities_per_second = 1000

[temporal.queue_options.anthropic-queue]
start_to_close_timeout = "0:05:00"
heartbeat_timeout      = "0:01:00"
max_task_queue_activities_per_second = 50
[temporal.queue_options.anthropic-queue.retry_policy_config]
initial_interval     = "0:00:01"
backoff_coefficient  = 2.0
maximum_interval     = "0:01:00"
maximum_attempts     = 5
# IMPORTANT: overlay uses ``_extra`` (additive), not the baseline field name.
non_retryable_error_types_extra = ["AnthropicAuthError"]

[temporal.queue_options.img-gen-gpu-queue]
start_to_close_timeout = "0:30:00"     # GPU calls take longer
heartbeat_timeout      = "0:02:00"
max_task_queue_activities_per_second = 10

# Rare per-handle override for ONE long-context model on the anthropic queue
[temporal.worker_config.activity_queues.act_llm_gen_text.handle_options."claude-4.6-opus"]
start_to_close_timeout = "0:15:00"

Resolution — three layers, last-wins for scalars

baseline
worker_config.default_activity_start_to_close_timeout
worker_config.default_activity_heartbeat_timeout
worker_config.retry_policy_config

↓ overlay

queue_options[resolved_queue]
Per-queue timeouts, retry overlay, queue-level rate cap.

↓ overlay

activity_queues[activity].handle_options[handle]
Per-handle timeout / retry override for a single model variant.

↓ result

✓

DispatchOptions
task_queue · start_to_close_timeout · schedule_to_close_timeout · schedule_to_start_timeout · heartbeat_timeout · retry_policy

The additive rule for non_retryable_error_types

Scalars (timeouts, attempts, intervals) are last-wins. non_retryable_error_types is additive — every layer can only add to the no-retry list, never remove from it. The schema enforces this asymmetry by splitting the retry-policy model in two:

Layer	Class	Field
Baseline only	`RetryPolicyConfig`	`non_retryable_error_types: list[str]` — the main list
Queue / handle overlay	`RetryPolicyConfigOverlay`	`non_retryable_error_types_extra: list[str]` — additive

Pydantic's extra="forbid" rejects the wrong field on each layer, so a config like queue_options.foo.retry_policy_config.non_retryable_error_types = [...] fails at load time with a clear validation error — you can't accidentally bypass the additive rule.

The resolver in code

WorkerConfig.resolve_dispatch(activity_name, routing_key, queue_options_by_queue, is_traced) -> DispatchOptions

@dataclass
class DispatchOptions:
    task_queue: str | None                         # None ⇒ omit kwarg, ride workflow's queue
    start_to_close_timeout: timedelta
    retry_policy: RetryPolicy
    schedule_to_close_timeout: timedelta | None = None
    schedule_to_start_timeout: timedelta | None = None
    heartbeat_timeout: timedelta | None = None

    def to_execute_kwargs(self) -> dict[str, Any]:
        # Splat into workflow.execute_activity(...) — omits keys whose value is None
        # so the Temporal SDK's own defaults kick in rather than receiving None.
        ...

queue_options without activity_queues — the asymmetry. If activity_queues is empty (no routing topology) but queue_options[default_task_queue] exists, the resolver still applies that queue's overlay. This lets single-queue deployments tune timeouts/retry/rate without opting into the routing model. Dispatch returns task_queue=None (activity rides the workflow's own queue) but the queue's options still flow through.

5. The call-site shape

Every workflow.execute_activity(...) in ContentGeneratorInWorkflow splats the resolver output. One uniform pattern across LLM, image-gen, jinja2, and extract calls:

async def make_llm_text(self, job_metadata, llm_setting_main, llm_prompt_for_text):
    worker_config = get_config().temporal.worker_config
    dispatch_kwargs = worker_config.resolve_dispatch(
        activity_name=act_llm_gen_text.__name__,
        routing_key=llm_assignment.llm_handle,                             # for by_handle / handle_options
        queue_options_by_queue=get_config().temporal.queue_options,
        is_traced=get_config().temporal.temporal_config.temporal_log_config.is_dispatch_resolution_traced,
    ).to_execute_kwargs()
    return await workflow.execute_activity(
        act_llm_gen_text,
        arg=llm_assignment,
        summary=build_activity_summary("LLM text", job_metadata,
                                       extras={"model": llm_assignment.llm_handle}),
        **dispatch_kwargs,    # task_queue + start_to_close + schedule_to_close + heartbeat + retry_policy
    )

6. Layer C — worker-runtime profiles

A single worker process selects one named profile via pipelex worker --profile <name>. The profile bundles every Worker(...) constructor knob.

[temporal.worker_runtime_profiles]
default_profile = "balanced"

[temporal.worker_runtime_profiles.profiles.balanced]
tuning_mode = "explicit"
max_cached_workflows               = 1000
max_concurrent_workflow_tasks      = 100
max_concurrent_activities          = 100
max_concurrent_local_activities    = 100
max_concurrent_workflow_task_polls = 5
max_concurrent_activity_task_polls = 5
max_activities_per_second          = 1000          # worker-local rate cap
sticky_queue_schedule_to_start_timeout = "0:00:10"
max_heartbeat_throttle_interval        = "0:01:00"
default_heartbeat_throttle_interval    = "0:00:30"
graceful_shutdown_timeout              = "0:00:30"

# Router profile — workflow-only, no activities at all.
[temporal.worker_runtime_profiles.profiles.router]
tuning_mode = "explicit"
max_cached_workflows               = 2000
max_concurrent_workflow_tasks      = 200
max_concurrent_activities          = 0             # 👈 ge=0 allows workflow-only workers
max_concurrent_local_activities    = 0
max_concurrent_workflow_task_polls = 10
max_concurrent_activity_task_polls = 0
max_activities_per_second          = 1
sticky_queue_schedule_to_start_timeout = "0:00:10"
max_heartbeat_throttle_interval        = "0:01:00"
default_heartbeat_throttle_interval    = "0:00:30"
graceful_shutdown_timeout              = "0:00:30"

# GPU profile — small slot count, long heartbeats
[temporal.worker_runtime_profiles.profiles.gpu]
tuning_mode = "explicit"
max_cached_workflows               = 100
max_concurrent_workflow_tasks      = 10
max_concurrent_activities          = 4             # GPU memory-bound
max_concurrent_local_activities    = 0
max_concurrent_workflow_task_polls = 2
max_concurrent_activity_task_polls = 2
max_activities_per_second          = 10
sticky_queue_schedule_to_start_timeout = "0:00:30"
max_heartbeat_throttle_interval        = "0:05:00"
default_heartbeat_throttle_interval    = "0:01:00"
graceful_shutdown_timeout              = "0:01:00"

Boot a specialized worker:

$ pipelex worker --task-queue img-gen-gpu-queue --profile gpu --scope runner-img-gen
$ pipelex worker --task-queue temporal_task_queue   --profile router  --scope router
$ pipelex worker --task-queue anthropic-queue       --profile balanced --scope runner-llm

Scopes (specialized worker scopes from the same diff): runner-llm, runner-img-gen, runner-extract, runner-jinja2. Each scope registers only its activities. act_render_page_views is registered under both runner-img-gen and runner-extract (belt-and-suspenders for the two paths that need it).

7. Strict CLI validation

pipelex worker --task-queue X now fast-fails if X isn't declared anywhere — default_task_queue, any activity_queues entry, or any queue_options key. With a Levenshtein "did you mean?" suggestion.

$ pipelex worker --task-queue antrhopic-queue
WorkerTaskQueueUnknownError: task queue 'antrhopic-queue' is not declared in config.
  Known queues: ['anthropic-queue', 'img-gen-gpu-queue', 'openai-queue',
                 'temporal_task_queue', 'extract-default-queue', 'extract-azure-queue']
  Did you mean: 'anthropic-queue'?

This is the strict counterpart to the lenient WARN at config-load on activity_queues entries that name unknown queues. Routing config can be sloppy in one direction (warn); the worker boot picks one queue and that one must exist (fail).

8. Dispatch resolution tracing

Flip a config flag to see exactly which layer each scalar came from on every activity call:

[temporal.temporal_config.temporal_log_config]
is_dispatch_resolution_traced = true

INFO  dispatch  act_llm_gen_text  routing_key=claude-4.6-opus
        task_queue=anthropic-queue            (activity_queues.by_handle)
        start_to_close_timeout=0:15:00        (handle_options)
        heartbeat_timeout=0:01:00             (queue_options)
        retry.maximum_attempts=5              (queue_options)
        retry.non_retryable=['AnthropicAuthError','OAIBillingError']  (baseline + queue_extra)

Off by default — verbose; turn it on when debugging mis-tuned timeouts or a queue that shouldn't be receiving traffic.

9. Workflow timeout vs activity timeout — separated

Two distinct budgets, set in distinct places:

workflow_execution_timeout on WorkerConfig bounds the workflow as a whole.
default_activity_start_to_close_timeout on WorkerConfig (ships at 10 min) is the baseline budget on each activity call.
queue_options[<queue>].start_to_close_timeout overrides the activity budget per queue.
handle_options[<handle>].start_to_close_timeout overrides it for one specific handle when the queue baseline doesn't fit.

Each kind of activity ends up with a timeout shaped to its real cost — a jinja2 render fails fast, a slow PDF extract gets the headroom it actually needs, and the workflow's own deadline isn't spent by a single in-flight activity.

10. Rate caps live in two distinct places

Knob	Where	Semantics
`max_task_queue_activities_per_second`	`[temporal.queue_options.<queue>]`	Cluster-wide cap on this queue. Conveyed to the Temporal server by every worker on the queue. Latest value wins.
`max_activities_per_second`	`[temporal.worker_runtime_profiles.profiles.<name>]`	Worker-local cap on this process. Caps what one worker pulls regardless of cluster headroom.

The shipping default [temporal.queue_options.temporal_task_queue] sets max_task_queue_activities_per_second = 1000. Deployments using non-default queue names should add their own [temporal.queue_options.<queue>] entries with the cap appropriate for that backend pool.

11. The whole picture, in one example

Say someone calls a pipe whose router needs an Anthropic Opus completion. Here's the trace from submit to the wire:

// inside WfPipeRouter: content_generator.make_llm_text(...)
// llm_assignment.llm_handle resolves to "claude-4.6-opus"
worker_config.resolve_dispatch("act_llm_gen_text", routing_key="claude-4.6-opus", queue_options_by_queue=...)
// step 1: resolve_queue → activity_queues["act_llm_gen_text"].by_handle["claude-4.6-opus"] = "anthropic-queue"
// step 2: overlay queue_options["anthropic-queue"] → start_to_close=5m, heartbeat=1m, retry: 5 attempts, +AnthropicAuthError
// step 3: overlay activity_queues["act_llm_gen_text"].handle_options["claude-4.6-opus"] → start_to_close=15m (longer context)
// step 4: non_retryable composed additively: baseline + queue._extra + handle._extra
→ DispatchOptions(task_queue="anthropic-queue", start_to_close_timeout=15m, heartbeat_timeout=1m, retry_policy=<5 attempts, no-retry: [...]>)
workflow.execute_activity(act_llm_gen_text, **dispatch_kwargs, summary="LLM text · pipe=translate_doc · model=claude-4.6-opus")
→ Temporal server routes the task to a worker polling anthropic-queue with the balanced profile.

12. Where to look

Config models: pipelex/temporal/config_temporal.py — WorkerConfig, QueueOptions, HandleOptions, ActivityRouteConfig, WorkerRuntimeProfile, DispatchOptions
The two resolvers: WorkerConfig.resolve_queue and WorkerConfig.resolve_dispatch in the same file
Call sites that splat the resolver output: pipelex/temporal/tprl_content_generation/content_generator_in_workflow.py — every make_* method
Defaults: pipelex/pipelex.toml under [temporal.worker_config], [temporal.queue_options.*], [temporal.worker_runtime_profiles.*]
Override invitations: .pipelex/pipelex.toml ships the same blocks commented out
Companion note: id-and-naming.html — Workflow IDs, child IDs, activity IDs, search attributes, summaries