> ## Documentation Index
> Fetch the complete documentation index at: https://usefused.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# harnest.task

> Public Python signatures, fields, and methods.

[All Python modules](https://usefused.com/docs/harnest/reference/api/overview.md) · [Import and compatibility guidance](https://usefused.com/docs/harnest/build/python-imports.md)

Use [`task`](#task) to decorate application work. It returns a [`TaskCallable`](#taskcallable) whose direct call runs inline and whose [`defer`](#taskcallable-defer) method submits queued work. The returned [`TaskHandle`](#taskhandle) exposes status, cancellation, and result access.

For discovery, storage setup, and durable waiting, see [Create and run Tasks](https://usefused.com/docs/harnest/build/queued-tasks.md), [Task and cron storage](https://usefused.com/docs/harnest/runtime/task-storage.md), and [Durable execution](https://usefused.com/docs/harnest/runtime/durable-execution.md).

## Task parameters

| Parameter | Contract |
| - | - |
| `task(function)` | Supports bare `@task` and configured `@task(...)`. The decorated function must have a docstring. |
| `queue="default"` | 1–64 characters. Starts with an ASCII letter; remaining characters may include ASCII letters, digits, `.`, `_`, `~`, and `-`. Set on the decorator, not on `.defer()`. |
| `max_retries=3` | Integer from 0 to 100. Set on the decorator. |
| `defer(*args, **kwargs)` | Bound against the task's original signature and serialized as strict JSON. Positional-only parameters and `*args` are unsupported in the decorated task. |
| `idempotency_key=None` | Optional nonblank text of at most 512 characters. Durable tools derive a replay-stable key when omitted. |
| `schedule_in=None` | Optional finite, non-negative delay in seconds. |

The names `idempotency_key` and `schedule_in` are options on `.defer()`; avoid using them as task input names. Arguments and results must be strict JSON values: no credentials, arbitrary Python objects, non-string object keys, or non-finite numbers. Workers may retry, so task side effects must be idempotent.

## Handle behavior

| Member | Contract |
| - | - |
| `id` | Opaque queue job identifier. |
| `task_name` | Compiler-owned registered task name. |
| `await status()` | Returns the owning runtime's job state. |
| `await cancel()` | Returns whether cancellation succeeded. |
| `await result()` | Returns the JSON-safe completed result. Unfinished work requires an async `@tool(durable=True)` and Harnest-owned checkpointer so the invocation can suspend and resume. |

Handles belong to the compiled runtime that returned them. There is no public handle reconstruction or job-listing API. Calling `.defer()` without an active compiled runtime raises `TaskUnavailableError`; calling the decorated function directly does not require queue storage.

`TaskStore` and `TaskRecord` are storage adapter contracts. `TaskDefinition` and `CompiledTask` describe compiler/runtime registrations. Use [Task and cron storage](https://usefused.com/docs/harnest/runtime/task-storage.md) to choose a production provider; `MemoryTaskStore` does not make work survive a process restart.

## Storage protocols

| Contract | Required operations |
| - | - |
| `TaskStore` | `start`, `close`, `enqueue_task`, `get_task`, `claim_tasks`, `renew_task_lease`, `finish_task`, `cancel_task` |
| `CronStore` | `start`, `close`, `create_cron`, `get_cron`, `list_crons`, `update_cron`, `delete_cron`, `list_due_crons`, `commit_cron_occurrence` |

## Exports

| Name | Kind |
| - | - |
| [`CompiledTask`](#compiledtask) | Class |
| [`MemoryTaskStore`](#memorytaskstore) | Class |
| [`PostgresTaskStore`](#postgrestaskstore) | Class |
| [`TaskCallable`](#taskcallable) | Class |
| [`TaskDefinition`](#taskdefinition) | Class |
| [`TaskHandle`](#taskhandle) | Class |
| [`TaskRecord`](#taskrecord) | Class |
| [`TaskStore`](#taskstore) | Class |
| [`TaskStoreConflictError`](#taskstoreconflicterror) | Class |
| [`TaskUnavailableError`](#taskunavailableerror) | Class |
| [`safe_task_result`](#safe_task_result) | Function |
| [`task`](#task) | Function |

## CompiledTask

One compiler-named task executable in the compiled application runtime.

```python theme={null}
CompiledTask(name: str, source: str, definition: TaskDefinition, authored: TaskCallable[Any]) -> None
```

Fields (`—` means no declared default):

| Field | Type | Default |
| - | - | - |
| `name` | `str` | `—` |
| `source` | `str` | `—` |
| `definition` | `TaskDefinition` | `—` |
| `authored` | `TaskCallable[Any]` | `—` |

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `name` | `str` | Required |
> | `source` | `str` | Required |
> | `definition` | `TaskDefinition` | Required |
> | `authored` | `TaskCallable[Any]` | Required |
>
> ```python theme={null}
> from harnest.task import CompiledTask
>
> task = CompiledTask(name=..., source=..., definition=..., authored=...)
> ```

### CompiledTask.function

Return the original callable only at the trusted worker boundary.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import CompiledTask
>
> task: CompiledTask
> value = task.function
> ```

### CompiledTask.queue

Return the authored queue without exposing the definition object.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import CompiledTask
>
> task: CompiledTask
> value = task.queue
> ```

### CompiledTask.max\_retries

Return the bounded retry count used during native registration.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import CompiledTask
>
> task: CompiledTask
> value = task.max_retries
> ```

## MemoryTaskStore

Implement atomic contracts in one process; records do not survive restart.

```python theme={null}
MemoryTaskStore() -> None
```

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store = MemoryTaskStore()
> ```

### MemoryTaskStore.start

```python theme={null}
async MemoryTaskStore.start() -> None
```

Satisfy lifecycle ownership without opening external resources.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> await store.start()
> ```

### MemoryTaskStore.close

```python theme={null}
async MemoryTaskStore.close() -> None
```

Release no resources; in-memory data lives only with this instance.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> await store.close()
> ```

### MemoryTaskStore.enqueue\_task

```python theme={null}
async MemoryTaskStore.enqueue_task(record: TaskRecord) -> TaskRecord
```

Commit an isolated task snapshot and its deduplication identity.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `TaskRecord` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> record = await store.enqueue_task(record=...)
> ```

### MemoryTaskStore.get\_task

```python theme={null}
async MemoryTaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None
```

Read one detached snapshot through the caller's ownership scope.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `user_id` | `str \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> record = await store.get_task(application_id=..., job_id=...)
> ```

### MemoryTaskStore.list\_task\_metadata

```python theme={null}
async MemoryTaskStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]
```

Return a bounded owner-scoped metadata page without copying private payloads.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.list_task_metadata(application_id=..., user_id=...)
> ```

### MemoryTaskStore.claim\_tasks

```python theme={null}
async MemoryTaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]
```

Claim due work, reclaim expired leases and retire exhausted attempts.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `queues` | `tuple[str, ...]` | Required |
> | `now` | `float` | Required |
> | `lease_seconds` | `float` | Required |
> | `limit` | `int` | `1` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.claim_tasks(
>     application_id=...,
>     queues=...,
>     now=...,
>     lease_seconds=...,
> )
> ```

### MemoryTaskStore.renew\_task\_lease

```python theme={null}
async MemoryTaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool
```

Extend only a lease that still grants execution ownership.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `lease_token` | `str` | Required |
> | `now` | `float` | Required |
> | `lease_seconds` | `float` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.renew_task_lease(
>     application_id=...,
>     job_id=...,
>     lease_token=...,
>     now=...,
>     lease_seconds=...,
> )
> ```

### MemoryTaskStore.finish\_task

```python theme={null}
async MemoryTaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool
```

Commit an outcome only while the exact execution lease remains valid.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `lease_token` | `str` | Required |
> | `now` | `float` | Required |
> | `status` | `str` | Required |
> | `result` | `Any` | `None` |
> | `failure_code` | `str \| None` | `None` |
> | `retry_at` | `float \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.finish_task(
>     application_id=...,
>     job_id=...,
>     lease_token=...,
>     now=...,
>     status=...,
> )
> ```

### MemoryTaskStore.cancel\_task

```python theme={null}
async MemoryTaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool
```

Cancel one owner's active work and revoke any in-flight attempt.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `now` | `float` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)
> ```

### MemoryTaskStore.create\_cron

```python theme={null}
async MemoryTaskStore.create_cron(record: CronRecord) -> CronRecord
```

Create a schedule or return an exact keyed retry without reactivation.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `CronRecord` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> record = await store.create_cron(record=...)
> ```

### MemoryTaskStore.get\_cron

```python theme={null}
async MemoryTaskStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None
```

Read a detached schedule through both ownership predicates.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `schedule_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)
> ```

### MemoryTaskStore.list\_crons

```python theme={null}
async MemoryTaskStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]
```

Return one bounded page from this in-memory reference collection.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.list_crons(application_id=..., user_id=...)
> ```

### MemoryTaskStore.list\_cron\_metadata

```python theme={null}
async MemoryTaskStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]
```

Project an owner-scoped schedule page without copying task arguments.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.list_cron_metadata(application_id=..., user_id=...)
> ```

### MemoryTaskStore.update\_cron

```python theme={null}
async MemoryTaskStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord
```

Replace allowed fields only at the current scoped revision.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `CronRecord` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> record = await store.update_cron(record=..., expected_revision=...)
> ```

### MemoryTaskStore.delete\_cron

```python theme={null}
async MemoryTaskStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool
```

Remove one owner's schedule and its reusable authored key.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `schedule_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)
> ```

### MemoryTaskStore.list\_due\_crons

```python theme={null}
async MemoryTaskStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]
```

Read active due schedules with a stable timestamp-and-ID cursor.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `now` | `float` | Required |
> | `after` | `tuple[float, str] \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> result = await store.list_due_crons(application_id=..., now=...)
> ```

### MemoryTaskStore.commit\_cron\_occurrence

```python theme={null}
async MemoryTaskStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None
```

Use one lock for revision check, task insert and schedule advancement.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `schedule_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
> | `due_at` | `float` | Required |
> | `next_run_at` | `float` | Required |
> | `task` | `TaskRecord` | Required |
>
> ```python theme={null}
> from harnest.task import MemoryTaskStore
>
> store: MemoryTaskStore
> record = await store.commit_cron_occurrence(
>     application_id=...,
>     user_id=...,
>     schedule_id=...,
>     expected_revision=...,
>     due_at=...,
>     next_run_at=...,
>     task=...,
> )
> ```

## PostgresTaskStore

Persist jobs and schedules with transactional claims and fenced leases.

`_pool` supports composing this provider with the combined session store;
externally supplied pools remain owned by their caller.

```python theme={null}
PostgresTaskStore(
    dsn: str,
    *,
    pool_options: Mapping[str, Any] | None = None,
    setup_schema: bool = True,
) -> None
```

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `dsn` | `str` | Required |
> | `pool_options` | `Mapping[str, Any] \| None` | `None` |
> | `setup_schema` | `bool` | `True` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store = PostgresTaskStore(dsn=...)
> ```

### PostgresTaskStore.start

```python theme={null}
async PostgresTaskStore.start() -> None
```

Open one pool and install or validate the durable schema.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> await store.start()
> ```

### PostgresTaskStore.close

```python theme={null}
async PostgresTaskStore.close() -> None
```

Close only the pool opened by this instance.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> await store.close()
> ```

### PostgresTaskStore.create\_cron

```python theme={null}
async PostgresTaskStore.create_cron(record: CronRecord) -> CronRecord
```

Create a schedule once without reactivating an existing cancelled key.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `CronRecord` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> record = await store.create_cron(record=...)
> ```

### PostgresTaskStore.get\_cron

```python theme={null}
async PostgresTaskStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None
```

Read a schedule only within its application and owner scope.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `schedule_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)
> ```

### PostgresTaskStore.list\_crons

```python theme={null}
async PostgresTaskStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]
```

Apply owner filtering and keyset pagination in the database.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.list_crons(application_id=..., user_id=...)
> ```

### PostgresTaskStore.update\_cron

```python theme={null}
async PostgresTaskStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord
```

Replace one schedule conditionally, preserving its immutable identity.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `CronRecord` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> record = await store.update_cron(record=..., expected_revision=...)
> ```

### PostgresTaskStore.delete\_cron

```python theme={null}
async PostgresTaskStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool
```

Delete a schedule using the same owner predicate as reads.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `schedule_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)
> ```

### PostgresTaskStore.list\_due\_crons

```python theme={null}
async PostgresTaskStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]
```

Page due schedules in database order without loading inactive rows.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `now` | `float` | Required |
> | `after` | `tuple[float, str] \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.list_due_crons(application_id=..., now=...)
> ```

### PostgresTaskStore.commit\_cron\_occurrence

```python theme={null}
async PostgresTaskStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None
```

Atomically enqueue an occurrence and advance its schedule revision.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `schedule_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
> | `due_at` | `float` | Required |
> | `next_run_at` | `float` | Required |
> | `task` | `TaskRecord` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> record = await store.commit_cron_occurrence(
>     application_id=...,
>     user_id=...,
>     schedule_id=...,
>     expected_revision=...,
>     due_at=...,
>     next_run_at=...,
>     task=...,
> )
> ```

### PostgresTaskStore.enqueue\_task

```python theme={null}
async PostgresTaskStore.enqueue_task(record: TaskRecord) -> TaskRecord
```

Persist an immutable task identity, returning exact idempotent retries.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `TaskRecord` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> record = await store.enqueue_task(record=...)
> ```

### PostgresTaskStore.get\_task

```python theme={null}
async PostgresTaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None
```

Read a scoped job; omitted user scope is reserved for trusted workers.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `user_id` | `str \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> record = await store.get_task(application_id=..., job_id=...)
> ```

### PostgresTaskStore.claim\_tasks

```python theme={null}
async PostgresTaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]
```

Claim disjoint batches and retire exhausted attempts after crashes.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `queues` | `tuple[str, ...]` | Required |
> | `now` | `float` | Required |
> | `lease_seconds` | `float` | Required |
> | `limit` | `int` | `1` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.claim_tasks(
>     application_id=...,
>     queues=...,
>     now=...,
>     lease_seconds=...,
> )
> ```

### PostgresTaskStore.renew\_task\_lease

```python theme={null}
async PostgresTaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool
```

Renew only an unexpired attempt still owned by the calling worker.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `lease_token` | `str` | Required |
> | `now` | `float` | Required |
> | `lease_seconds` | `float` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.renew_task_lease(
>     application_id=...,
>     job_id=...,
>     lease_token=...,
>     now=...,
>     lease_seconds=...,
> )
> ```

### PostgresTaskStore.finish\_task

```python theme={null}
async PostgresTaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool
```

Fence completion and retries; terminal transitions erase private input.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `lease_token` | `str` | Required |
> | `now` | `float` | Required |
> | `status` | `str` | Required |
> | `result` | `Any` | `None` |
> | `failure_code` | `str \| None` | `None` |
> | `retry_at` | `float \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.finish_task(
>     application_id=...,
>     job_id=...,
>     lease_token=...,
>     now=...,
>     status=...,
> )
> ```

### PostgresTaskStore.cancel\_task

```python theme={null}
async PostgresTaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool
```

Cancel an owned nonterminal job and revoke any running attempt.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `now` | `float` | Required |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)
> ```

### PostgresTaskStore.list\_task\_metadata

```python theme={null}
async PostgresTaskStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]
```

Push owner, cursor, projection, ordering, and limit into one database query.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.list_task_metadata(application_id=..., user_id=...)
> ```

### PostgresTaskStore.list\_cron\_metadata

```python theme={null}
async PostgresTaskStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]
```

Read a bounded metadata projection with owner filtering in one SQL query.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.task import PostgresTaskStore
>
> store: PostgresTaskStore
> result = await store.list_cron_metadata(application_id=..., user_id=...)
> ```

## TaskCallable

Preserve direct invocation while adding an asynchronous defer boundary.

```python theme={null}
TaskCallable(definition: TaskDefinition) -> None
```

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `definition` | `TaskDefinition` | Required |
>
> ```python theme={null}
> from harnest.task import TaskCallable
>
> task_callable = TaskCallable(definition=...)
> ```

### TaskCallable.**call**

```python theme={null}
TaskCallable.__call__(*args: Any, **kwargs: Any) -> Any
```

Execute inline when the author deliberately calls the task itself.

> [accordion] Parameters and example
>
> | Parameter | Type | Default | Description |
> | - | - | - | - |
> | `*args` | `Any` | Optional | Positional arguments for the task function. |
> | `**kwargs` | `Any` | Optional | Keyword arguments for the task function. |
>
> ```python theme={null}
> # build_report is your @task function
> report = await build_report(account_id="acct_123")  # runs inline, no queue
> ```

### TaskCallable.defer

```python theme={null}
async TaskCallable.defer(
    *args: Any,
    idempotency_key: str | None = None,
    schedule_in: float | None = None,
    **kwargs: Any,
) -> TaskHandle
```

Commit safe arguments to the active compiled task runtime.

> [accordion] Parameters and example
>
> | Parameter | Type | Default | Description |
> | - | - | - | - |
> | `*args` | `Any` | Optional | Positional arguments for the task function. |
> | `idempotency_key` | `str \| None` | `None` | Optional key, at most 512 characters, that deduplicates submissions. |
> | `schedule_in` | `float \| None` | `None` | Optional delay in seconds before the task can run. |
> | `**kwargs` | `Any` | Optional | Keyword arguments for the task function, stored as strict JSON. |
>
> ```python theme={null}
> # build_report is your @task function
> handle = await build_report.defer(account_id="acct_123")
> ```

## TaskDefinition

Authoring metadata retained on a decorated function.

```python theme={null}
TaskDefinition(function: Callable[..., Any], queue: str, max_retries: int) -> None
```

Fields (`—` means no declared default):

| Field | Type | Default |
| - | - | - |
| `function` | `Callable[..., Any]` | `—` |
| `queue` | `str` | `—` |
| `max_retries` | `int` | `—` |

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `function` | `Callable[..., Any]` | Required |
> | `queue` | `str` | Required |
> | `max_retries` | `int` | Required |
>
> ```python theme={null}
> from harnest.task import TaskDefinition
>
> definition = TaskDefinition(function=..., queue=..., max_retries=...)
> ```

## TaskHandle

Opaque queue handle returned after a job is committed.

Obtain this object from the owning Harnest API; its constructor includes runtime bindings.

Fields (`—` means no declared default):

| Field | Type | Default |
| - | - | - |
| `id` | `str` | `—` |
| `task_name` | `str` | `—` |

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskHandle
>
> handle: TaskHandle = await build_report.defer(account_id="acct_123")
> ```

### TaskHandle.status

```python theme={null}
async TaskHandle.status() -> str
```

Read the native job state through the owning runtime.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskHandle
>
> handle: TaskHandle
> result = await handle.status()
> ```

### TaskHandle.cancel

```python theme={null}
async TaskHandle.cancel() -> bool
```

Cancel queued work without exposing the backend application.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskHandle
>
> handle: TaskHandle
> result = await handle.cancel()
> ```

### TaskHandle.result

```python theme={null}
async TaskHandle.result() -> Any
```

Return completed work or suspend through a durable managed tool.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskHandle
>
> handle: TaskHandle
> result = await handle.result()
> ```

## TaskRecord

Private persisted job, including reconstructable invocation capabilities.

Epoch timestamps are UTC seconds. Attempts count claims, starting at one;
`max_retries` permits that many attempts after the first. Providers retain
an immutable enqueue fingerprint after clearing terminal private payloads.

```python theme={null}
TaskRecord(
    job_id: str,
    application_id: str,
    user_id: str,
    task_name: str,
    queue: str,
    arguments: Mapping[str, Any],
    invocation: Mapping[str, Any] | None = None,
    agent_permissions: tuple[str, ...] | None = None,
    trigger: str = 'agent',
    status: str = 'pending',
    scheduled_at: float = 0.0,
    max_retries: int = 3,
    attempt: int = 0,
    lease_token: str | None = None,
    lease_expires_at: float | None = None,
    result: Any = None,
    failure_code: str | None = None,
    idempotency_key: str | None = None,
    created_at: float = 0.0,
    updated_at: float = 0.0,
    cron_schedule_id: str | None = None,
) -> None
```

Fields (`—` means no declared default):

| Field | Type | Default |
| - | - | - |
| `job_id` | `str` | `—` |
| `application_id` | `str` | `—` |
| `user_id` | `str` | `—` |
| `task_name` | `str` | `—` |
| `queue` | `str` | `—` |
| `arguments` | `Mapping[str, Any]` | `—` |
| `invocation` | `Mapping[str, Any] \| None` | `None` |
| `agent_permissions` | `tuple[str, ...] \| None` | `None` |
| `trigger` | `str` | `'agent'` |
| `status` | `str` | `'pending'` |
| `scheduled_at` | `float` | `0.0` |
| `max_retries` | `int` | `3` |
| `attempt` | `int` | `0` |
| `lease_token` | `str \| None` | `None` |
| `lease_expires_at` | `float \| None` | `None` |
| `result` | `Any` | `None` |
| `failure_code` | `str \| None` | `None` |
| `idempotency_key` | `str \| None` | `None` |
| `created_at` | `float` | `0.0` |
| `updated_at` | `float` | `0.0` |
| `cron_schedule_id` | `str \| None` | `None` |

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `job_id` | `str` | Required |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_name` | `str` | Required |
> | `queue` | `str` | Required |
> | `arguments` | `Mapping[str, Any]` | Required |
> | `invocation` | `Mapping[str, Any] \| None` | `None` |
> | `agent_permissions` | `tuple[str, ...] \| None` | `None` |
> | `trigger` | `str` | `'agent'` |
> | `status` | `str` | `'pending'` |
> | `scheduled_at` | `float` | `0.0` |
> | `max_retries` | `int` | `3` |
> | `attempt` | `int` | `0` |
> | `lease_token` | `str \| None` | `None` |
> | `lease_expires_at` | `float \| None` | `None` |
> | `result` | `Any` | `None` |
> | `failure_code` | `str \| None` | `None` |
> | `idempotency_key` | `str \| None` | `None` |
> | `created_at` | `float` | `0.0` |
> | `updated_at` | `float` | `0.0` |
> | `cron_schedule_id` | `str \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import TaskRecord
>
> record = TaskRecord(
>     job_id=...,
>     application_id=...,
>     user_id=...,
>     task_name=...,
>     queue=...,
>     arguments=...,
> )
> ```

## TaskStore

Atomic operations required for at-least-once task execution.

All selection, scope predicates, ordering, and limits execute inside the
datastore. A provider must durably commit before returning success. Expired
lease holders cannot renew or finish work. Terminal transitions clear
arguments, invocation and permissions while retaining the result/failure.

```python theme={null}
TaskStore(*args, **kwargs)
```

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> class MyTaskStore(TaskStore):
>     ...
> ```

### TaskStore.start

```python theme={null}
async TaskStore.start() -> None
```

Open owned resources and initialize the provider's durable schema.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> await store.start()
> ```

### TaskStore.close

```python theme={null}
async TaskStore.close() -> None
```

Release owned resources after workers and dispatchers have stopped.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> await store.close()
> ```

### TaskStore.enqueue\_task

```python theme={null}
async TaskStore.enqueue_task(record: TaskRecord) -> TaskRecord
```

Atomically persist payload and job, or return an identical submission.

Idempotency is scoped by application, user, task name and optional key.
Conflicting definitions raise TaskStoreConflictError. Persist a separate
fingerprint so terminal payload cleanup cannot defeat deduplication.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `TaskRecord` | Required |
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> record = await store.enqueue_task(record=...)
> ```

### TaskStore.get\_task

```python theme={null}
async TaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None
```

Read one scoped job; omitted user is a trusted worker-only lookup.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `user_id` | `str \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> record = await store.get_task(application_id=..., job_id=...)
> ```

### TaskStore.claim\_tasks

```python theme={null}
async TaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]
```

Atomically claim due pending or expired running jobs in queue scope.

Return a bounded eligible batch with best-effort earliest-due ordering;
concurrent workers and lease recovery do not guarantee strict FIFO.
Each claim increments attempt and issues a fresh lease token.
Expired final attempts become failed with
task\_failed instead of remaining stranded or retrying indefinitely.
Empty queues selects no work. Limits are integers from 1 to 1000 and
apply to returned claims.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `queues` | `tuple[str, ...]` | Required |
> | `now` | `float` | Required |
> | `lease_seconds` | `float` | Required |
> | `limit` | `int` | `1` |
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> result = await store.claim_tasks(
>     application_id=...,
>     queues=...,
>     now=...,
>     lease_seconds=...,
> )
> ```

### TaskStore.renew\_task\_lease

```python theme={null}
async TaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool
```

Extend a running, unexpired matching lease; stale owners return false.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `lease_token` | `str` | Required |
> | `now` | `float` | Required |
> | `lease_seconds` | `float` | Required |
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> result = await store.renew_task_lease(
>     application_id=...,
>     job_id=...,
>     lease_token=...,
>     now=...,
>     lease_seconds=...,
> )
> ```

### TaskStore.finish\_task

```python theme={null}
async TaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool
```

Fence a completion, failure or retry with a matching unexpired lease.

Status is completed, failed, or pending. Pending requires retry\_at and
remaining retries; it preserves payload. All transitions clear leases.
Stale owners return false and must never overwrite another attempt.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `lease_token` | `str` | Required |
> | `now` | `float` | Required |
> | `status` | `str` | Required |
> | `result` | `Any` | `None` |
> | `failure_code` | `str \| None` | `None` |
> | `retry_at` | `float \| None` | `None` |
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> result = await store.finish_task(
>     application_id=...,
>     job_id=...,
>     lease_token=...,
>     now=...,
>     status=...,
> )
> ```

### TaskStore.cancel\_task

```python theme={null}
async TaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool
```

Atomically cancel pending/running work, invalidate leases and scrub.

Unknown, foreign and terminal jobs return false. A running effect may
already have happened; cancellation prevents subsequent state commits.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `job_id` | `str` | Required |
> | `now` | `float` | Required |
>
> ```python theme={null}
> from harnest.task import TaskStore
>
> store: TaskStore
> result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)
> ```

## TaskStoreConflictError

A stable job identity or idempotency key has a different definition.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskStoreConflictError
>
> try:
>     ...
> except TaskStoreConflictError as error:
>     ...
> ```

## TaskUnavailableError

A task was deferred without its compiled runtime being active.

> [accordion] Example
>
> ```python theme={null}
> from harnest.task import TaskUnavailableError
>
> try:
>     ...
> except TaskUnavailableError as error:
>     ...
> ```

## safe\_task\_result

Normalize a task result to the strict JSON values safe for persistence.

```python theme={null}
safe_task_result(value: Any) -> Any
```

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `value` | `Any` | Required |
>
> ```python theme={null}
> from harnest.task import safe_task_result
>
> result = safe_task_result(value=...)
> ```

## task

Mark a callable for queue, scheduling, and retry execution.

Deferred execution uses Harnest workers and the application's configured
`lifecycle.storage.tasks` provider. Durable storage makes queued work
recoverable across restarts; importing this API requires no queue library.
Direct calls remain ordinary Python calls; `defer` is queue execution and
never promises restoration of a suspended Python frame.

```python theme={null}
task(function: F | None = None, *, queue: str = 'default', max_retries: int = 3) -> Any
```

> [accordion] Parameters and example
>
> | Parameter | Type | Default | Description |
> | - | - | - | - |
> | `function` | `F \| None` | `None` | The decorated function. Filled in by bare `@task`; omit it when passing options. Needs a docstring and JSON-safe arguments. |
> | `queue` | `str` | `'default'` | Queue name, 1–64 characters, starting with a letter. Set it here, not on `.defer()`. |
> | `max_retries` | `int` | `3` | Retries after a failed run, from 0 to 100. |
>
> ```python theme={null}
> from harnest.task import task
>
> @task(queue="reports", max_retries=3)
> async def build_report(account_id: str) -> dict:
>     """Build one account report."""
>     return {"account_id": account_id}
> ```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.