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:

Create lib/task_storage.py

lib/task_storage.py

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"])

Create lifecycle/task_storage.py

lifecycle/task_storage.py

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: Follow TaskStore and, for schedules, CronStore. Their method references define required operations, parameters, and results. 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: a crash after an external effect can cause a retry. Make effects idempotent. Cancellation blocks later commits but cannot undo effects or forcibly stop synchronous code. Durable handle.result() waits also require a persistent Harnest-owned checkpointer. Task storage alone is insufficient. Queue ordering is best-effort 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.

Old queue data and continuation references are not migrated automatically. Keep the old database for recovery, and drain work before removing, renaming, or changing a task’s queue.

API reference

Reference: harnest.testing.

Optional development inspection

The Playground Work view uses optional list_task_metadata and list_cron_metadata methods. Existing TaskStore and CronStore implementations remain compatible without them: users can look up work by ID, cancel tasks, and manage known dynamic schedules. Both methods are async and accept keyword-only application_id, user_id, after=None, and limit=100 (1–100). Apply owner and cursor predicates, ordering, projection, and limit inside the datastore. Task IDs and schedule IDs are exclusive cursors sorted lexicographically. Return {"items": [...], "after": next_id_or_none}; task pages also include "indexing": false. A provider rebuilding an index can return an empty task page with "indexing": true until it can produce a complete listing. Task items contain only job_id, task_name, queue, status, scheduled_at, attempt, max_retries, created_at, updated_at, and failure_code. Cron items contain only schedule_id, key, expression, task_name, status, timezone, next_run_at, revision, max_runs, max_consecutive_failures, run_count, consecutive_failures, and read_only. Set read_only for keys beginning with static:. Never return task arguments, results, invocation snapshots, credentials, permissions, or lease tokens in metadata listings. Built-in memory, PostgreSQL, and Redis providers implement these queries. PostgreSQL adds owner indexes during schema setup; deployments using setup_schema=False should provision matching (application_id, user_id, job_id) and (application_id, user_id, schedule_id) indexes. Redis maintains the task index in the enqueue transaction and incrementally indexes older records when the Work view first lists them.

Overview

Get started

Harnest Studio

Guides

API reference

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

API reference

Optional development inspection

​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

​API reference

​Optional development inspection

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

API reference

Optional development inspection