API

Workflow Continuations

Grow sealed workflows through compiled, hash-pinned member policies.

Ordinary follow-ups create detached child jobs. A workflow continuation creates the child as a member of the parent’s already sealed workflow, without reopening the graph to arbitrary mutation.

This surface first shipped in 0.1.0a25 and remains in 0.1.0a36, Protocol document revision 1.0.17, and SQL contract 0.2.5+ (migrations 0016–0017).

1. Declare member edges

Mark the parent’s allowed target and the emitted follow-up as workflow members:

tasks.py

from taskq import Complete, Followup, FollowupTarget, Task, TaskRegistry


async def discover(payload: DiscoverInput) -> Complete:
    return Complete(
        result={"found": True},
        followups=(
            Followup(
                step="enrich",
                queue="enrichment",
                job_type="listing.enrich",
                payload={"listing_id": payload.listing_id},
                workflow_member=True,
            ),
        ),
    )


ENRICH = Task(
    name="listing.enrich",
    queue="enrichment",
    input_model=EnrichInput,
    output_model=EnrichOutput,
    handler=enrich,
)

DISCOVER = Task(
    name="listing.discover",
    queue="discovery",
    input_model=DiscoverInput,
    output_model=DiscoverOutput,
    followup_targets=(
        FollowupTarget(
            queue=ENRICH.queue,
            job_type=ENRICH.name,
            workflow_member=True,
            continuation_revision="1",
        ),
    ),
    handler=discover,
)

registry = TaskRegistry((DISCOVER, ENRICH))
policy = registry.compile_continuation_policy((DISCOVER,))

The policy compiler validates the reachable member graph, canonicalizes it, and produces a SHA-256 continuation_policy_hash. Change the continuation revision when the allowed durable edge set changes.

2. Bind the workflow to the policy

producer.py

from taskq import WorkflowKind

workflow = await tq.create_workflow(
    "listing-import:batch-42",
    WorkflowKind.DAG,
    declared_queues=policy.reachable_queues,
    actor="import-api",
    member_limit=10_000,
    continuation_policy_hash=policy.continuation_policy_hash,
)

await tq.enqueue(
    DISCOVER,
    {"listing_id": "listing-123"},
    workflow_id=workflow.workflow_id,
    step_key="root",
)
await tq.seal_workflow(workflow.workflow_id, actor="import-api")

member_limit is a durable cap for the entire workflow. The workflow records the policy hash at creation; a producer cannot attach one later.

3. Run policy-aware workers

Workers must advertise and retain every policy hash they can execute:

worker.py

from taskq import WorkerService, WorkerServiceOptions

service = WorkerService(
    runner_transport,
    registry,
    worker_id="worker-1",
    options=WorkerServiceOptions(
        queues=policy.reachable_queues,
        listen=False,
    ),
    continuation_policies=(policy,),
)
await service.run()

The stock taskq worker run CLI loads a registry but does not currently accept compiled continuation policies. Use a small programmatic worker entry point like the one above for continuation-enabled queues.

At claim time the worker advertises supported hashes. It will not claim a workflow member pinned to an unknown policy. If an unsupported claim reaches a worker through a custom transport, the worker releases it without invoking the handler and begins a soft stop.

Runtime guarantees

The claimed policy—not the current live registry—is the authority for that workflow’s member edges.
Member settlement and child insertion are atomic.
TaskQ derives the child step_key and an idempotency key in the reserved chain: namespace from the parent job id plus Followup.step.
Producer-supplied idempotency keys beginning with chain: are rejected.
A detached follow-up may still be emitted from the same handler by omitting workflow_member=True.
A workflow cannot exceed its durable member limit.

Retain older compiled policies while any workflow pinned to them can still run. A deploy that removes a live policy hash strands those members until a compatible worker returns.

When to use it

Use continuations when the next workflow member is discovered only while a current member executes. Use an ordinary sealed DAG when all members and edges are known before sealing. Use detached follow-ups when the child should not affect workflow completion.

Durable admission

Reserve / finish / cancel two-phase admission for producers that must not lose intent.

Standalone Scheduler

Bind the database target, apply paused manifests, and run one safe scheduler clock.