Durable Background Jobs

Persist background jobs so in-flight work survives a process restart, with opt-in deliver-back replay

from praisonaiagents import Agent
from praisonaiagents.tools.subagent_tool import create_subagent_tool

subagent = create_subagent_tool()  # spawn_subagent + subagent_result

# Background subagent jobs are persisted out of the box to
# <runs_dir>/background_jobs.db — status survives a restart.
agent = Agent(
    name="Researcher",
    instructions="Spawn a background subagent for long-running research.",
    tools=[subagent["function"], subagent["result_tool"]["function"]],
)

agent.start("Summarise the last week of arXiv AI papers in the background")

Durable background jobs persist every state transition so a crash mid-job no longer silently drops the work — orphans become a queryable LOST state, and a manager that calls reconcile_on_start(redeliver=...) replays undelivered results.

Durability is on by default: the shared manager returned by get_job_manager() — the same one the chat /tasks command and background subagent deliver-backs use — automatically persists to <runs_dir>/background_jobs.db, so job status survives a restart. Replaying undelivered results requires a host that calls reconcile_on_start(redeliver=...) (see the warning below). Opt out with PRAISONAI_BACKGROUND_JOB_STORE=0 for the prior pure in-memory behaviour.

Three ways to reach durability — pick the row that matches your setup.

Quick Start

1. You already have durability

   The shared manager returned by get_job_manager() is already backed by SqliteBackgroundJobStore and has already reconciled orphaned jobs on first construction — zero code, zero setup:

   from praisonaiagents.background.job_manager import get_job_manager
   
   manager = get_job_manager()  # already backed by SqliteBackgroundJobStore
                                # already ran reconcile_on_start() once

2. Enable persistence on a custom manager

   Building your own BackgroundJobManager? Pass store=SqliteBackgroundJobStore() (or any store implementing the BackgroundJobStore protocol). Every job transition is now persisted:

   from praisonaiagents.background import SqliteBackgroundJobStore
   from praisonaiagents.background.job_manager import BackgroundJobManager
   
   manager = BackgroundJobManager(store=SqliteBackgroundJobStore())

3. Reconcile on startup

   get_job_manager() already ran this once. On a custom manager, call reconcile_on_start() once at boot, after wiring the deliver-back handler:

   counts = manager.reconcile_on_start(redeliver=on_job_complete)
   # {"lost": 0, "redelivered": 0, "rehydrated": 0} on a clean start
   print(f"jobs reconciled: {counts}")

4. Query a job after a restart

   Status lookups work after a restart — even before reconcile_on_start() runs, get_status falls through to the store:

   from praisonaiagents.background.job_manager import JobStatus
   
   status = manager.get_status(job_id)  # e.g. JobStatus.LOST

5. Handle LOST jobs

   Inspect orphaned jobs and decide whether to retry or surface them to the user — abandoned work is never auto-re-run:

   from praisonaiagents.background.job_manager import JobStatus
   
   lost = manager.list_jobs(status=JobStatus.LOST)
   for job_id, info in lost.items():
       print(f"orphaned: {job_id} (origin={info.origin})")

How It Works

reconcile_on_start() reads store.list_unreconciled() and follows two paths: orphaned jobs become LOST; completed-but-undelivered jobs are replayed.

Every persisted job is re-hydrated into the in-memory map so get_status(job_id) keeps working after a restart. redeliver=None is safe — undelivered jobs are re-hydrated but not delivered.

State machine

LOST is a new terminal state reached only via restart reconciliation.

The BackgroundJobStore Protocol

Implement this @runtime_checkable protocol to plug in your own durable store. A concrete SQLite implementation — SqliteBackgroundJobStore — now ships in core (praisonaiagents.background) and backs get_job_manager() by default; you only implement the protocol yourself for a non-SQLite backend.

from typing import Protocol, runtime_checkable, Optional, List
from praisonaiagents.background.job_manager import JobInfo

@runtime_checkable
class BackgroundJobStore(Protocol):
    def upsert(self, job: JobInfo) -> None: ...
    def get(self, job_id: str) -> Optional[JobInfo]: ...
    def list_unreconciled(self) -> List[JobInfo]: ...

An optional delete_older_than(ts) -> int method, when present on a store, is called best-effort by cleanup_completed() with the same max_age policy so the durable record stays bounded. It is not part of the protocol, so existing custom stores keep working unchanged — add it when you want pruning.

Built-in SqliteBackgroundJobStore

SqliteBackgroundJobStore is the concrete, stdlib-only store that backs get_job_manager() by default. Import it, or omit db_path to use the default location:

from praisonaiagents.background import SqliteBackgroundJobStore

# Default location: <runs_dir>/background_jobs.db (honours PRAISONAI_HOME)
store = SqliteBackgroundJobStore()

# In-memory database for tests
test_store = SqliteBackgroundJobStore(db_path=":memory:")

Schema

A single background_jobs table with an index on status:

Index: idx_background_jobs_status on status.

Thread-safety & durability posture

A single re-entrant lock guards one shared connection opened check_same_thread=False — safe to share across the runner’s worker threads. Durability matches its siblings SQLiteRunLedger and SqliteEventLog: WAL journal mode, busy_timeout=5000, synchronous=NORMAL.

Configuration

Environment variable

Reconciliation counts

reconcile_on_start() returns a counts dict — use it for a single startup log line:

counts = manager.reconcile_on_start(redeliver=on_job_complete)
# {"lost": 2, "redelivered": 1, "rehydrated": 3}

Best Practices

Related