Workflows and schedules

Sealed workflows

Create a workflow, attach dependency edges, then seal it. Until sealed, the graph is mutable; after seal, jobs move through blocked → queued as dependencies succeed.

from taskq import WorkflowKind

workflow = await tq.create_workflow(
    "publish-wave:2026-08-02",
    WorkflowKind.DAG,
    declared_queues=("publishing",),
    actor="api",
)

prepare = await tq.enqueue(
    PREPARE,
    {"article_id": "article-123"},
    workflow_id=workflow.workflow_id,
    step_key="prepare",
)
await tq.enqueue(
    PUBLISH,
    {"article_id": "article-123"},
    workflow_id=workflow.workflow_id,
    step_key="publish",
    depends_on=(prepare.job_id,),
)
await tq.seal_workflow(workflow.workflow_id, actor="api")

Workflow keys are caller-chosen durable identities. Declare every queue the workflow may use, give each member a unique step_key, and add dependencies by job id before sealing. A sealed DAG is immutable.

Use sealed workflows when you need multi-job dependency graphs. Prefer atomic follow-ups for simple parent→child fan-out on settle.

Shipped in migrations 0009_workflows.sql (ADR-026).

Database-time schedules

Recurring definitions live in Postgres and use database time. The standalone taskq scheduler run process evaluates recurrence and durably enqueues ordinary jobs; application workers execute their handlers.

one scheduler clock -> queued TaskQ job -> application worker -> handler

The scheduler never imports application task registries or routers. See Standalone Scheduler for target binding, process placement, activation, monitoring, and rollback.

Source-owned manifests

Use YAML manifests for deployment-owned recurring work:

version: 1
namespace: myapp
source: deployment
schedules:
  nightly-cleanup:
    display_name: Nightly cleanup
    task: cleanup.run
    queue: maintenance
    cron: "0 3 * * *"
    timezone: UTC
    catchup: fire_once
    overlap: forbid
    max_lateness_seconds: 3600
    state: paused

taskq --context staging schedule manifest plan schedules.yaml -o json
taskq --context staging schedule manifest apply schedules.yaml \
  --plan-digest "$PLAN_DIGEST" -o json

plan is read-only. apply is source-scoped, version-checked, and does not prune missing keys. Retire one removed key explicitly with taskq --context staging --yes schedule manifest retire.

Low-level operator API

Use put_schedule when schedules are genuinely dynamic application data rather than deployment-owned desired state:

from taskq import (
    ScheduleCatchupPolicy,
    ScheduleCronRecurrence,
    ScheduleDefinition,
    ScheduleJobTarget,
)
from taskq.sql.transport import SqlTaskqTransport

definition = ScheduleDefinition(
    target=ScheduleJobTarget(
        queue="maintenance",
        job_type="cleanup.run",
        payload={},
    ),
    recurrence=ScheduleCronRecurrence(
        expression="0 3 * * *",
        timezone="UTC",
    ),
    catchup_policy=ScheduleCatchupPolicy.FIRE_ONCE,
    max_catchup=1,
    paused=True,
)

operator = SqlTaskqTransport.from_dsn(
    OPERATOR_DATABASE_URL,
    expected_environment="staging",
)
try:
    await operator.put_schedule(
        "nightly.cleanup",
        definition,
        actor="ops",
    )
finally:
    await operator.aclose()

put_schedule is an operator-transport operation, not a producer convenience method on TaskQ. Use a restricted taskq_operator connection; never give operator rights to producers or runners.

The base schedule contract shipped in migration 0010. Migrations 0019 and 0020 add target attestation, the standalone scheduler, durable decisions, manifest ownership, overlap/lateness policy, and auto-pause.

Finite projections

Observer-facing ready / running / finished / workflow pages are finite projections over the job ledger (migrations 0011–0013, ADR-029). Bounded worker presence is activated in 0014–0015 (ADR-033). Native workflow-member continuations are added and activated in 0016–0017.