> ## 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.

# Create and run Tasks

> Define, defer, inspect, cancel, and await retryable application work.

Use `@task` for application-owned queue work. A task is not model-visible; an Agent Tool decides when to defer it.

| Call                            | Behavior                                                 |
| ------------------------------- | -------------------------------------------------------- |
| `await build_report(...)`       | Runs an async task body as an ordinary local Python call |
| `await build_report.defer(...)` | Commits queue work and returns `TaskHandle`              |
| `await handle.result()`         | Returns a terminal result or suspends a durable tool     |

## Define and call a task

From the agent root, create a compilable task starter:

```bash theme={null}
harnest add task build-report
```

Use `--project <agent-root>` when running the command from elsewhere. The generated file is suitable when no other module imports the task; use the shared-library pattern below when tools also import it.

When a tool needs the task callable, define it once in `lib/` and re-export it from `tasks/`:

<CodeGroup>
  ```python lib/report_tasks.py 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, "status": "ready"}
  ```

  ```python tasks/build_report.py theme={null}
  from harnest.lib.report_tasks import build_report
  ```

  ```python tools/request_report.py theme={null}
  from harnest.lib.report_tasks import build_report
  from harnest.agent import tool


  @tool(durable=True)
  async def request_report(account_id: str) -> dict:
      """Queue a report and return it when ready."""

      handle = await build_report.defer(account_id=account_id)
      return await handle.result()
  ```
</CodeGroup>

If nothing imports the task, you can define it directly in `tasks/<name>.py`. The file must export exactly one same-named `@task` callable.

| Authoring rule       | Contract                                 |
| -------------------- | ---------------------------------------- |
| Location             | Flat root `tasks/<name>.py`              |
| Export               | Exactly one same-named `@task` callable  |
| Documentation        | A docstring is required                  |
| Body                 | Synchronous or asynchronous              |
| Arguments and result | Strict JSON values only                  |
| Deferred signature   | No positional-only parameters or `*args` |

## Control a job

The returned `TaskHandle` supports `await handle.status()`, `await handle.cancel()`, and `await handle.result()`. A handle belongs to the active compiled runtime; there is no public API to reconstruct or list handles by ID.

| Option             | Purpose                          |
| ------------------ | -------------------------------- |
| `queue=`           | Select a worker queue            |
| `max_retries=`     | Set queue retry policy           |
| `schedule_in=`     | Delay submission by seconds      |
| `idempotency_key=` | Deduplicate a logical submission |

Inside `@tool(durable=True)`, Harnest derives a replay-stable idempotency key when you omit one. Keep task effects idempotent because workers can retry.

Use `await task.defer(...)` without `handle.result()` for fire-and-forget work. Await a result only from an async `@tool(durable=True)` when the agent must resume after the Task finishes. An unfinished result requires a Harnest-owned checkpointer.

## Runtime requirements

| Requirement          | Behavior                                                                                                             |
| -------------------- | -------------------------------------------------------------------------------------------------------------------- |
| Queue backend        | Harnest workers with an explicit `lifecycle.storage.tasks` provider; no fallback                                     |
| Installation         | `harnest_postgres` and `harnest_redis` are bundled with Harnest; declare custom adapter dependencies in your project |
| Database             | PostgreSQL, Redis, or a custom provider selected by your Lifecycle factory                                           |
| Payloads and results | JSON-safe values only                                                                                                |
| Credentials          | Resolved at execution; never serialized into task state                                                              |
| Runtime principal    | Permission names and invocation identity are persisted separately; the principal object and credentials are not      |
| Defer boundary       | Available only while the compiled Harnest runtime is active                                                          |

Configure [Task and cron storage](/docs/harnest/runtime/task-storage) before serving queued work. That guide shows how to set up a database connection in a shared factory and provide its configuration through the environment. Every replica can run a worker against the same provider. Run these commands from your agent folder after configuring storage:

```bash theme={null}
harnest env sync .
harnest test .
harnest serve .
```

The selected provider initializes its storage during runtime startup. Compiling a task definition or calling its body directly does not require task storage; starting queued execution does. Missing storage produces an actionable setup error. Calling `.defer()` from an ordinary imported module without an active compiled runtime raises `TaskUnavailableError`; call the task body directly for isolated unit tests.

## Start a fresh agent session

A task can invoke the compiled root agent without an HTTP call:

```python tasks/review_report.py theme={null}
from harnest import context
from harnest.task import task


@task(max_retries=3)
async def review_report(report_id: str) -> dict:
    """Start a fresh agent session to review one completed report."""

    session = await context.agent.create_session(
        state={"report_id": report_id},
        key=report_id,
    )
    response = await session.invoke("Review the completed report")
    return response.as_dict()
```

`key` is optional. When supplied, Harnest derives the same opaque child session
and invocation identities on a task retry. User-deferred tasks inherit the
calling user and public metadata. Static schedules use Harnest's automation
identity; dynamic schedules run as their creating user with fresh context and
empty grants. Calls still pass through storage, plugins, lifecycle, tools, and the
selected framework runtime.

When `.defer()` runs under an [Agent Runtime Principal](/docs/harnest/runtime/agent-runtime-principals), Harnest stores only its permission names and reconstructs a fresh principal for each worker attempt. Agent calls made by the Task inherit that restriction unless trusted task code passes a narrower principal. A Task deferred without an active principal preserves compatibility behavior.

Durable external waits return a typed `in_progress` response. Human approvals
and client tools fail closed because their continuation is process-local after
the task returns.

<Warning>
  An unfinished `handle.result()` requires an async `@tool(durable=True)` and a Harnest-owned checkpointer. Harnest resumes framework execution; it does not restore a Python stack.
</Warning>

See [Durable execution](/docs/harnest/runtime/durable-execution) for cross-replica resume behavior.

## Test Task behavior

Call the decorated function directly to unit-test its business logic without a database. Exercise `.defer()`, retries, cancellation, and durable resume through a compiled smoke test with your chosen persistent provider so the real queue and checkpoint boundaries run together. Custom providers can use the [shared conformance suite](/docs/harnest/runtime/task-storage#test-a-custom-adapter).