Tasks

Tasks are the units of work in a v3 application. They replace v2's ActivitiesInterface and @activity.defn with a single @task decorator on App methods.

Defining a Task

Decorate any async method on your App subclass with @task. Each task method takes exactly one Input and returns exactly one Output -- both Pydantic models.

from application_sdk.app import App, task
from application_sdk.contracts import Input, Output

class FetchInput(Input):
    connection_id: str

class FetchOutput(Output):
    rows_fetched: int

class MyConnector(App):
    @task
    async def fetch_data(self, input: FetchInput) -> FetchOutput:
        # ... fetch logic
        return FetchOutput(rows_fetched=42)

Under the hood, @task applies the Temporal @activity.defn decorator, registers the method in the TaskRegistry, and wires heartbeating. You never import from temporalio directly.

Typed Input/Output Contracts

Every task boundary is a typed contract. The SDK validates contract fields at class definition time -- forbidden types raise PayloadSafetyError before your app starts.

Forbidden types: Any, bytes, bytearray, unbounded list[T], unbounded dict[K, V].

Safe alternatives:

Need	Use
Bounded list	`Annotated[list[str], MaxItems(1000)]`
Large binary / file data	`FileReference` (stored in object store)
Enum values	`SerializableEnum`

from typing import Annotated
from application_sdk.contracts import Input, Output
from application_sdk.contracts import MaxItems, FileReference

class ProcessInput(Input):
    items: Annotated[list[str], MaxItems(5000)]

class ProcessOutput(Output):
    results: FileReference  # large data stored in object store

SQL extraction inputs normalise the connector form's filter keys before task inputs are built: for example exclude_table_regex is routed into ExtractionInput.temp_table_regex, which is what SqlApp hands to its fetch tasks. See Excluding tables and views by name.

Reporting Failure as Data

Most tasks report failure by raising -- Temporal marks the activity failed and retries it per the task's retry policy. A few must not: when retrying is itself the damage, the task catches the exception and returns a structured failure on its Output, and the caller (run()) decides what to do.

SqlApp.prime_sql_auth is the SDK's worked example. It issues one SELECT 1 to warm the source's auth cache before the parallel extract burst. If the source rejects the credential, an activity-level retry just stacks the source's failed_login_attempts counter -- accelerating the very lockout the probe exists to prevent. So the task is declared retry_max_attempts=1 and returns PrimeAuthOutput(success=False, failure=...) instead of raising.

The typed failure contract

failure is a FailureDetails -- the SDK's wire envelope for a classified error (see common.md). It carries routing (category, audience, retryable, code), a message, an optional suggested_action, and per-error evidence.

SqlApp.run() consumes it by rebuilding the typed error and raising it, so the workflow fails with the verdict the task reached -- routing, evidence and remediation hint intact:

prime_result = await self.prime_sql_auth(task_input)
if not prime_result.success:
    raise self._classify_prime_failure(prime_result)

Classify where the exception is, not where the strings are. Only serialisable fields cross an activity boundary, and BaseSQLClient.get_results rewraps every driver exception in one wrapper class with one fixed message -- so a DNS failure, a TLS error and a credential rejection arrive at the caller looking identical. A caller classifying from those strings routes all three to INTERNAL / APP_OWNER, telling Atlan to investigate a customer credential problem. Classifying inside the task, where the live __cause__ chain still exists, is what makes AuthError / USER reach the right first-responder.

Apply the same shape to your own task: classify against the live exception, put the verdict on the output, and let the caller rebuild it.

Legacy string fields

PrimeAuthOutput also carries error_type and error_message. They are a fallback, not the contract -- read failure first:

Field	Semantics
`failure`	Authoritative typed verdict. `None` on success, and on outputs from an older SDK.
`error_type`	Class name of the root cause (innermost `__cause__`), not the SDK wrapper around it.
`error_message`	Message of the root cause, secret-redacted and truncated to 500 chars.

_classify_prime_failure falls back to matching those two strings only when failure is absent -- an activity result written by an older SDK and replayed from Temporal history, or a PrimeAuthOutput a connector built by hand.

Redaction is the producer's job

A SQL driver message routinely embeds the whole connection string, password included, and everything on an output is persisted in Temporal history and in logs. The SDK therefore redacts at every boundary that writes to the wire: at capture (error_message), when serialising a typed verdict (message, suggested_action, and recursively through nested evidence values), and again when rebuilding an error from a replayed envelope -- history may predate the redaction, and the envelope's denylist rejects secret-named evidence keys only, never a secret sitting in a value.

If your task returns failure as data, redact before it leaves the task:

from application_sdk.errors import redact_secrets, safe_traceback

@task(retry_max_attempts=1)
async def probe(self, input: ProbeInput) -> ProbeOutput:
    try:
        await self._connect(input)
    except Exception as exc:
        # Never exc_info=True here -- the traceback renders the driver
        # message, connection string and all, verbatim into the log.
        logger.error("probe failed\n%s", safe_traceback(exc))
        return ProbeOutput(success=False, error_message=redact_secrets(str(exc)))
    return ProbeOutput(success=True)

Evidence keys are constrained too: FailureDetails rejects secret-named keys (api_key, db_password, *_token, …) outright. prime_sql_auth degrades such a verdict by dropping the offending keys and keeping the typed routing -- never by falling back to a string-matched guess, which would silently invert the connector's own verdict.

Timeouts and Auto-Heartbeating

Configure timeouts and heartbeating as keyword arguments on @task:

class MyConnector(App):
    @task(
        timeout_seconds=3600,            # Temporal kills the task after 1 hour
        heartbeat_timeout_seconds=60,    # Temporal kills the task if no heartbeat in 60s
        auto_heartbeat_seconds=10,       # framework sends a heartbeat every 10s
    )
    async def long_running(self, input: MyInput) -> MyOutput:
        # heartbeats run automatically in a background asyncio loop
        ...

There is no @auto_heartbeater decorator in v3. Heartbeating is declarative.

Process-wide timeout defaults via env vars

When no explicit value is passed to @task, the framework reads these env vars at process startup:

Env var	Default	Controls
`ATLAN_START_TO_CLOSE_TIMEOUT_SECONDS`	`86400` (24 h)	`timeout_seconds` — Temporal kills the activity attempt after this many seconds
`ATLAN_HEARTBEAT_TIMEOUT_SECONDS`	`60`	`heartbeat_timeout_seconds` — Temporal restarts the activity if no heartbeat is received within this window
`ATLAN_SCHEDULE_TO_CLOSE_TIMEOUT_SECONDS`	unset	`schedule_to_close_seconds` — ceiling on the task's total time across every retry. Unset (or `0`) leaves the retry product unbounded
`ATLAN_PROGRESS_WATCHDOG`	`warn`	`progress_watchdog` — the stall watchdog's mode. `off` is a kill-switch that beats a per-task declaration
`ATLAN_MAX_NO_PROGRESS_SECONDS`	`900`	`max_no_progress_seconds` — how long an attempt may be silent before the watchdog reports a gap

Set these in atlan.yaml (or your deployment env) to apply a fleet-wide default without touching every @task decorator:

# atlan.yaml
env:
  - name: ATLAN_START_TO_CLOSE_TIMEOUT_SECONDS
    value: "1800"   # 30 min default for all tasks in this app
  - name: ATLAN_HEARTBEAT_TIMEOUT_SECONDS
    value: "120"    # 2 min heartbeat window

Explicit @task(timeout_seconds=...) values always take precedence over the env vars.

The three knobs

Timeout-shaped settings are easy to confuse, so it is worth being explicit about which question each one answers:

Knob	Default	Answers	Tune it?
`heartbeat_timeout_seconds`	60s	"Is anything beating at all?" — crash, OOM kill, node loss, a fully blocked event loop	No
`max_no_progress_seconds`	900s	"How long may this attempt be silent before it counts as stalled?" — wedged-but-alive	Rarely
`timeout_seconds` (`start_to_close`)	a 24h backstop	"What is the absolute ceiling on one attempt?"	No — it is a backstop, not a budget

There is a fourth setting, progress_watchdog (off / warn / enforce, defaulting to warn fleet-wide), which decides what the watchdog does with a gap rather than how long a gap may be.

max_no_progress_seconds is not the beat interval (that is auto_heartbeat_seconds, and it is unconditional) and not a duration budget — it measures the gap between progress signals, so a nine-hour task that writes a batch every thirty seconds never approaches it. See Progress and Stalls for the full picture: what counts as progress, when you need holding_progress(), and how to read the report the watchdog produces for your app.

All three bound one attempt. The ceiling on a task's total time across every attempt is a separate question, with its own answer below.

Bounding the retry product

timeout_seconds bounds one attempt. The retry policy multiplies it, and StartToClose is a retryable failure in Temporal, so a wedged task can spend the whole product:

worst case per dispatch = retry_max_attempts x timeout_seconds

At the defaults — three attempts against the 24-hour backstop — that is 72 hours. The SDK leaves that product unbounded on purpose (ADR-0018 → Bounding total time) — picking a total duration up front is the guess the whole design removes — so the ceiling is a declaration, available three ways:

from application_sdk.execution.retry import retry_product_seconds

class MyConnector(App):
    # 1. Bound the total. A ceiling below timeout_seconds caps the attempt too,
    #    so this works while inheriting a generous per-attempt backstop.
    @task(schedule_to_close_seconds=7200)
    async def extract(self, input: MyInput) -> MyOutput: ...

    # 2. Bound at exactly what the retry policy already implies — makes today's
    #    worst case explicit and enforced, and changes nothing about one attempt.
    @task(
        timeout_seconds=3600,
        retry_max_attempts=3,
        schedule_to_close_seconds=retry_product_seconds(3600, 3),
    )
    async def transform(self, input: MyInput) -> MyOutput: ...

    # 3. Drop the multiplier instead — the right shape for a backstop-class task
    #    where a retry cannot rescue anything.
    @task(retry_max_attempts=1)
    async def publish(self, input: MyInput) -> MyOutput: ...

Set ATLAN_SCHEDULE_TO_CLOSE_TIMEOUT_SECONDS to apply a ceiling to every task in the app without touching a decorator. Two caveats worth stating:

A ceiling smaller than two attempts makes the retry policy cosmetic — the second attempt cannot start inside the window. retry_product_seconds exists so that arithmetic is explicit rather than accidental.
The ceiling bounds one dispatch, which is what the retry policy spends. Worker-eviction re-dispatches (pod SIGTERM: KEDA scale-down, spot reclaim, rolling deploy) each get a fresh window by design, and are bounded separately by ATLAN_WORKER_EVICTION_MAX_RETRIES.

The run-length SLA alert

A ceiling on the retry product does not bound a run that keeps making some progress: every progress signal re-arms the stall watchdog, so it never fires, and no per-attempt timeout is ever reached. For that shape the SDK raises a duration alert rather than a duration kill — it cannot know how long a healthy run takes against a tenant it has never seen, but a run past the length its own operator declared is worth a human look.

Every executing task attempt compares its run's age against the SLA on the same tick as its heartbeat. Past the SLA it logs one WARNING and records task.run_length_over_sla (a histogram of the run's age, re-asserted once a minute for as long as the run stays over), so the alert keeps firing while the run does and resolves once it ends.

Env var	Default	Controls
`ATLAN_RUN_LENGTH_SLA_SECONDS`	`86400` (24 h)	How long a run may take before the alert is raised. `0` disables it

Nothing terminates the run. See Monitoring → Run length for the alert rule and the long-running-run runbook for what an operator does with one. An app whose healthy runs legitimately outlast 24 hours declares its own value; that declaration is the point, and it costs one env var rather than a code change.

Manual Heartbeats with Progress

For tasks that should resume from where they left off after a retry, send typed heartbeat details:

from application_sdk.contracts import HeartbeatDetails

class MyProgress(HeartbeatDetails):
    last_id: str
    records_done: int

class MyConnector(App):
    @task(heartbeat_timeout_seconds=60)
    async def process_batches(self, input: MyInput) -> MyOutput:
        prev = self.task_context.get_heartbeat_details(MyProgress)
        start_id = prev.last_id if prev else None

        for batch in get_batches(start_from=start_id):
            process(batch)
            self.task_context.heartbeat(
                MyProgress(last_id=batch.id, records_done=batch.count)
            )

self.heartbeat() / self.task_context.heartbeat() is safe to call from blocking code you offloaded with run_in_thread, including a plain-function RollingFileWriter on_chunk_complete: the SDK hands the beat back to the event loop.

The details survive a worker eviction too: when a pod is shut down mid-task (KEDA scale-down, spot reclaim, rolling deploy) the SDK re-dispatches the task as a new activity execution, and the last details the evicted attempt sent are carried across, so get_heartbeat_details() still returns them. Details that were never sent cannot be carried — beat at every safe resume point.

Two things bound the carry, both deliberate:

A beat with no details clears the checkpoint. self.heartbeat() with no arguments says this attempt has moved past whatever was carried in, so nothing is carried out of it. That is what you want when the work is done; it is not a keepalive. To beat without changing the checkpoint, pass the same details again, or rely on the automatic heartbeat.
Details must be encodable by the Temporal data converter. They ride to the next execution on the eviction failure itself, so anything that cannot be serialised is dropped with a WARNING rather than carried. Beat with a HeartbeatDetails model or plain JSON-native values — the eviction is still re-dispatched either way, but an unserialisable checkpoint is lost.

A manual beat also marks progress for the stall watchdog, under the label task.heartbeat — so a custom loop that beats once per iteration is observable and needs no hold. It is the third of the three progress mechanisms in Progress and Stalls.

Infrastructure Access via self.context

Inside a @task method, access infrastructure through self.context:

class MyConnector(App):
    @task
    async def fetch(self, input: FetchInput) -> FetchOutput:
        # Secret store
        api_key = await self.context.get_secret("my-api-key")

        # Credential resolution
        cred = await self.context.resolve_credential(input.credential_ref)
        ...

You do not create DaprClient instances or call SecretStore statics. Infrastructure is injected by the framework -- Dapr-backed in production, in-memory mocks in tests.

Blocking Sync Code

Prefer native async libraries wherever possible. For legacy sync code that cannot be rewritten, self.task_context.run_in_thread(fn, *args) offloads the call to a thread pool, preventing it from stalling the event loop and blocking heartbeats. See ADR-0010 for when this is appropriate and the required internal-timeout precautions.

Every offload is automatically wrapped in an unbounded progress hold, so a long blocking call is never read as a stall — see Progress and Stalls for how to give it a real bound instead.

Cancelling a blocking call at the driver

Cancelling the task that awaits run_in_thread hands the event loop back at once, but the worker thread runs on: Python cannot kill a thread. Until the call returns by itself it holds a slot in the shared sdk-blocking- pool (min(32, cpu + 4) wide), a database connection, and warehouse compute. Enough abandoned calls fill the pool, and every other offload queues behind them.

If the driver can stop the call from another thread, pass a CancelHandle and register the driver's cancel on it from inside the worker thread, at the moment the driver handle exists:

from application_sdk.execution.heartbeat import CancelHandle

handle = CancelHandle()

def _query() -> list[tuple[object, ...]]:
    cursor = connection.cursor()
    handle.set(cursor.cancel)  # the handle exists only here, in the thread
    cursor.execute(sql)
    return cursor.fetchall()

rows = await self.task_context.run_in_thread(handle.bind(_query))

Every run_in_thread call has a handle — the one bound with bind, or a fresh one — so you don't have to create one. Code anywhere inside the call reaches it with current_cancel_handle(), without the handle being passed down:

from application_sdk.execution.heartbeat import current_cancel_handle

def _query() -> list[tuple[object, ...]]:
    cursor = connection.cursor()
    handle = current_cancel_handle()
    if handle is not None:  # None outside run_in_thread
        handle.set(cursor.cancel)
    cursor.execute(sql)
    return cursor.fetchall()

rows = await self.task_context.run_in_thread(_query)

When the awaiting task is cancelled — by Temporal, asyncio.timeout or a failing TaskGroup sibling — run_in_thread calls handle.request() and re-raises CancelledError immediately.
The action runs at most once, on a small dedicated sdk-cancel- pool — never on the loop, and never behind the sdk-blocking- calls it is meant to stop.
If the cancel was requested before set() (a statement id known only after submit, say), set() fires the action straight away.
A failing action is logged at WARNING, never raised.
The request is scoped to that one call. The caller, and every other offloaded call, still read False.
Between steps of a loop, is_cancelled() (or handle.requested) reports the request, and raise_if_cancelled() raises asyncio.CancelledError. That is a BaseException, so a per-record except Exception does not swallow it.
For a nested scope inside one call — a helper opening its own connection while the caller's is mid-query — use with handle.registered(conn.interrupt):. It puts the previous action back when the block exits, so the outer query stays cancellable. A plain set() would leave the handle pointing at the closed inner connection.
The handle travels on the callable (handle.bind(func)), so every keyword argument still reaches func; no keyword name is reserved.

After a cancel, don't return the connection to a pool: invalidate it. BaseSQLClient already does this for its own read paths — see Clients.

!!! warning "Never stop or suspend a warehouse from a cancel action" A cancel stops this statement. It does not stop a warehouse the statement caused to start resuming — that start was triggered server-side — and stopping or suspending the warehouse would take it from every other user of it. Cancel actions must call statement- or cursor-level cancels only.

A client-side cancel is the second line of defence. The first is a server-side statement timeout set in the session (for example Snowflake's STATEMENT_TIMEOUT_IN_SECONDS and STATEMENT_QUEUED_TIMEOUT_IN_SECONDS), which bounds the statement even when the worker is gone.

FileReference: Passing Large Data Between Tasks

Temporal has a ~2 MB payload limit. Use FileReference for large data:

from application_sdk.contracts import FileReference

class FetchOutput(Output):
    results: FileReference  # automatically uploaded when output leaves the task

class ProcessInput(Input):
    results: FileReference  # automatically downloaded before the task runs

class MyConnector(App):
    @task
    async def fetch(self, input: FetchInput) -> FetchOutput:
        write_parquet(data, "/tmp/results.parquet")
        return FetchOutput(results=FileReference.from_local("/tmp/results.parquet"))

    @task
    async def process(self, input: ProcessInput) -> ProcessOutput:
        data = read_parquet(input.results.local_path)
        ...

    async def run(self, input: FetchInput) -> ProcessOutput:
        fetch_out = await self.fetch(input)
        return await self.process(ProcessInput(results=fetch_out.results))

FileReference handles task-to-task data passing within a run (automatic via the activity interceptor). For hand-off to Atlan system apps (publish, lineage, quality), call App.upload() explicitly from run() — see storage.md and file-reference.md for the two-store routing details.

`ExtractionOutput.transformed_files` is model-declared

A FileReference field on an entry point's boundary needs a declared shape — a warning today, an error in v4.0 (see Apps — Declaring artifact schemas). ExtractionOutput.transformed_files is the one you inherit rather than write, and it needs no artifactSchemas entry:

class MyExtractionOutput(ExtractionOutput):   # transformed_files comes with a declaration
    row_count: int = 0

The field carries the AssetArtifact marker, which says its declaration is the pyatlan_v9 Asset model. It is the SDK's field end to end — declared on ExtractionOutput, populated by SqlApp.run(), written by SqlApp._transform_entity — so no connector authors any part of it, and the only envelope a connector could hand-write for it would be a partial restatement of a 500-type model.

At the task boundary the interceptor therefore validates it against the whole Asset backbone (per-record decode plus .validate(), plus a referential/orphan pass) rather than against a field map, in an isolated child process. The marker resolves across the MRO, so narrowing the field or attaching your own Field(...) keeps the declaration. If your pkl contract already declares this field, delete that entry — it is ignored at runtime, because the model wins. See Entry points — AssetArtifact.

Auto-Discovery

You do not register tasks manually. When your App subclass is defined, App.__init_subclass__ scans it for @task methods and registers them in the TaskRegistry. The worker discovers all registered tasks at startup via create_worker().

Testing Tasks

Tasks can be tested without any Dapr sidecar or Temporal server by injecting in-memory infrastructure:

import pytest
from application_sdk.testing import MockSecretStore, MockStateStore
from application_sdk.infrastructure import InfrastructureContext, set_infrastructure

@pytest.fixture
def infra():
    ctx = InfrastructureContext(
        secret_store=MockSecretStore({"api-key": "test-secret"}),
        state_store=MockStateStore(),
    )
    set_infrastructure(ctx)
    return ctx

async def test_fetch_data(infra):
    connector = MyConnector()
    output = await connector.fetch_data(FetchInput(connection_id="test"))
    assert output.rows_fetched > 0

Navigation Menu

Repository navigation

tasks.md

tasks.md

Tasks

Defining a Task

Typed Input/Output Contracts

Reporting Failure as Data

The typed failure contract

Legacy string fields

Redaction is the producer's job

Timeouts and Auto-Heartbeating

Process-wide timeout defaults via env vars

The three knobs

Bounding the retry product

The run-length SLA alert

Manual Heartbeats with Progress

Infrastructure Access via self.context

Blocking Sync Code

Cancelling a blocking call at the driver

FileReference: Passing Large Data Between Tasks

`ExtractionOutput.transformed_files` is model-declared

Auto-Discovery

Testing Tasks

FilesExpand file tree

tasks.md

Latest commit

History

tasks.md

File metadata and controls

Tasks

Defining a Task

Typed Input/Output Contracts

Reporting Failure as Data

The typed failure contract

Legacy string fields

Redaction is the producer's job

Timeouts and Auto-Heartbeating

Process-wide timeout defaults via env vars

The three knobs

Bounding the retry product

The run-length SLA alert

Manual Heartbeats with Progress

Infrastructure Access via self.context

Blocking Sync Code

Cancelling a blocking call at the driver

FileReference: Passing Large Data Between Tasks

ExtractionOutput.transformed_files is model-declared

Auto-Discovery

Testing Tasks

`ExtractionOutput.transformed_files` is model-declared