Task and cron storage

Use PostgreSQL, Redis, or your own storage without changing your Tasks. Harnest runs the workers, retries, and schedules; your provider persists their state. Queued tasks require @lifecycle.storage.tasks. Add @lifecycle.storage.cron to the same factory for recurring work. There is no default database.

Choose and register a provider

harnest_postgres and harnest_redis ship with the Harnest runtime, not as separate PyPI distributions. Managed ADK and LangGraph environments include both drivers. From your agent folder, synchronize the matching runtime:

harnest env sync .

For direct Python development from source, install the driver you need:

python -m pip install '/path/to/harnest[postgres]'

python -m pip install '/path/to/harnest[redis]'

Use [postgres,redis] to install both drivers. Imports do not open connections. Register one shared factory in lifecycle/storage.py, replacing the corresponding existing storage factories:

import os

from harnest import lifecycle
from harnest_postgres import PostgresStore


@lifecycle.storage.sessions
@lifecycle.storage.checkpoints
@lifecycle.storage.tasks
@lifecycle.storage.cron
def storage() -> PostgresStore:
    """Share one managed PostgreSQL pool across durable storage roles."""
    return PostgresStore(os.environ["DATABASE_URL"])

import os

from harnest import lifecycle
from harnest_redis import RedisStore


@lifecycle.storage.sessions
@lifecycle.storage.checkpoints
@lifecycle.storage.tasks
@lifecycle.storage.cron
def storage() -> RedisStore:
    """Use Redis for conversation state, tasks, and recurring schedules."""
    return RedisStore(os.environ["REDIS_URL"])

Role	Configuration
Tasks and cron	Must share one instance so scheduling and enqueueing are atomic. Omit the cron decorator for tasks alone.
Sessions and checkpoints	Can share this provider or use a separate one.
Local tests	Use `harnest.task.MemoryTaskStore`; jobs disappear on process exit.

Harnest calls the factory, starts the provider, and closes it once. Remove unused decorators; keep any file that still provides sessions or checkpoints.

harnest test .
harnest serve .

PostgreSQL creates Harnest-owned task and cron tables. For Redis, configure persistence, replication, backups, and no eviction; a cache-only deployment is not durable. Task records do not inherit session/checkpoint TTLs. Start your database service before the agent. Workers run inside Harnest; no separate queue engine is needed. For reusable configuration, move the factory into lib/. Use this instead of the inline task/cron factory above:

import os

from harnest_redis import RedisStore


def create_store() -> RedisStore:
    """Read deployment configuration without opening a connection at import time."""
    return RedisStore(os.environ["TASK_STORAGE_URL"])

from harnest import lifecycle
from harnest.lib.task_storage import create_store


@lifecycle.storage.tasks
@lifecycle.storage.cron
def task_storage():
    """Share one managed connection across tasks and recurring schedules."""
    return create_store()

For PostgreSQL, substitute harnest_postgres.PostgresStore and a PostgreSQL URL. Declare custom adapter dependencies in your agent’s pyproject.toml. From your agent folder, point the factory at your database and start the service:

export TASK_STORAGE_URL='redis://127.0.0.1:6379/0'
harnest env sync .
harnest test .
harnest serve .

Match environment-variable names to your factory and inject production credentials through deployment secrets. Replicas must share the same application identity and persistent provider.

Write a custom database adapter

Implement the public structural contracts; subclassing a Harnest store is not required:

from harnest.task import TaskRecord, TaskStore, TaskStoreConflictError
from harnest.cron import CronRecord, CronStore, CronStoreConflictError

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`

Follow the protocol signatures and method docstrings. Records use UTC epoch seconds, JSON-safe payloads, and detached snapshots.

Guarantee	Provider responsibility
Enqueue	Persist payload and job atomically. Scope idempotency to application, user, task, and key; retain a fingerprint after payload cleanup.
Leases	Claim atomically with a fresh attempt token. Fence renewals and results with the matching, unexpired token.
Retries	Persist timing and attempt counts; terminalize exhausted crashed attempts.
Cancellation	Invalidate leases atomically; scrub arguments, invocation snapshots, and permissions at terminal transitions.
Isolation	Apply owner predicates, ordering, and bounded pagination in the database. `get_task(user_id=None)` is for trusted workers, not Agent Tools.
Cron	Revision-check edits; atomically commit each occurrence and advance its cursor through the same provider.

Test a custom adapter

Put this in your adapter package’s tests and run it against an isolated real database, not only a mock:

import unittest

from harnest.testing import TaskStoreConformanceMixin
from my_storage import MyStore


class MyStoreContractTests(TaskStoreConformanceMixin,
                           unittest.IsolatedAsyncioTestCase):
    async def make_store(self) -> MyStore:
        """Return a fresh facade over an isolated test database."""
        return MyStore(...)

python -m unittest discover -s tests -v

The suite checks isolation, concurrent claims, leases, deduplication, revisions, and atomic scheduling. Add database-specific crash, rollback, connection-loss, and restart tests. Verify deployment persistence separately.

Execution and recovery guarantees

Delivery is at least once, not exactly-once external effects. A worker may crash after sending an email but before committing its result. Keep side effects idempotent. Cancellation prevents later state commits and requests cooperative execution cancellation; it cannot undo an effect or forcibly stop a synchronous function already running in a thread. The provider runtime polls for work, renews attempt leases, and periodically reconciles retained results with durable continuation storage. Use a persistent Harnest-owned checkpointer for handle.result() to resume across restarts; persistent Task storage alone does not make a memory checkpointer durable. Queue ordering is best-effort, not strict FIFO across replicas and recovery. Cron cursors survive restart. Missed occurrences catch up in bounded batches; there is no skip/coalesce misfire option in this version. Cancelling a schedule before its atomic occurrence commit prevents that enqueue; already committed jobs continue. Removed static declarations are paused at the next deployment startup, while a retargeted declaration gets a new identity. Do not run old and new conflicting static declarations concurrently during a rolling deployment.

Move an existing application safely

The Procrastinate backend and automatic PostgreSQL fallback have been removed. If you used them, drain work on your currently deployed version before upgrading. The new runtime cannot process the old queue tables.

Back up your database and inventory recurring schedules for each owner.
Stop new submissions and pause recurring schedules on the old runtime.
Drain or explicitly cancel queued/running Tasks and resolve their waiting continuations before stopping the old workers.
Configure the new provider, preserving session/checkpoint storage when it contains history you need. Synchronize the runtime and recompile your agent. Start the new deployment and recreate dynamic schedules under their original user scopes. Static declarations reconcile automatically.
Verify task execution, recovery, and owner isolation before reopening traffic.

There is no automatic copy of old queue rows, schedule IDs, results, or continuation references. Keep the old database until its retention and recovery needs are satisfied. Likewise, drain work before renaming/removing a Task or changing its queue: workers subscribe to currently compiled queues, not every historical queue in the database.

Overview

Get started

Guides

Task and cron storage

Choose and register a provider

Write a custom database adapter

Test a custom adapter

Execution and recovery guarantees

Move an existing application safely

​Choose and register a provider

​Share connection configuration

​Write a custom database adapter

​Test a custom adapter

​Execution and recovery guarantees

​Move an existing application safely

Choose and register a provider

Share connection configuration

Write a custom database adapter

Test a custom adapter

Execution and recovery guarantees

Move an existing application safely