Durable Background Jobs

Persist background jobs so in-flight work and deliver-backs survive a process restart

from praisonaiagents import Agent

# Background subagent deliver-back — durable out of the box.
# If the process restarts mid-job, the deliver-back is replayed on
# the next boot from <PRAISONAI_HOME>/background_jobs.db.
agent = Agent(
    name="Researcher",
    instructions="Do long-running research and hand back a summary.",
)

result_id = agent.spawn_subagent(
    task="Summarise the last week of arXiv AI papers",
    background=True,
    deliver="chat",
)

Durable background jobs persist every state transition so a crash mid-job no longer silently drops the work or the promised deliver-back — orphans become a queryable LOST state and undelivered results are replayed on the next boot.

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 and reconciles on boot. 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]: ...

Method	Purpose
upsert(job)	Insert or update the persisted record, keyed by job_id
get(job_id)	Return the persisted JobInfo, or None if unknown
list_unreconciled()	Return orphaned (PENDING/RUNNING) jobs plus COMPLETED-with-origin jobs never marked delivered

The runner calls upsert() from its worker threads — your store implementation must be thread-safe.

---

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

Aspect	Detail
Import	from praisonaiagents.background import SqliteBackgroundJobStore (or from praisonaiagents.background.sqlite_store import SqliteBackgroundJobStore)
Constructor	SqliteBackgroundJobStore(db_path: Optional[str] = None)
Default DB path	<runs_dir>/background_jobs.db — under PRAISONAI_HOME, alongside the run ledger and event log
Tests	Pass db_path=":memory:"
Extra method	close() closes the underlying connection

Schema

A single background_jobs table with an index on status:

Column	Type	Notes
job_id	TEXT	Primary key
status	TEXT	pending / running / completed / failed / cancelled / lost
created_at	REAL	Spawn timestamp
started_at	REAL	Nullable
completed_at	REAL	Nullable
result	TEXT	JSON-serialised job result
error	TEXT	Nullable error string
origin	TEXT	JSON-serialised deliver-back context
delivered	INTEGER	0/1 — deliver-back fired

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

Symbol	Type	Default	Description
store	Optional[BackgroundJobStore]	None	Keyword-only. Enable persistence when supplied; pure in-memory when None. get_job_manager() supplies a SqliteBackgroundJobStore automatically
JobStatus.LOST	enum value	—	Terminal state for RUNNING/PENDING jobs interrupted by a crash
JobInfo.delivered	bool	False	Whether the deliver-back to origin has fired
reconcile_on_start	method	—	reconcile_on_start(redeliver=None) -> Dict[str, int]

Environment variable

Env var	Default	Purpose
PRAISONAI_BACKGROUND_JOB_STORE	1 (on)	Set to 0 / false to force the shared get_job_manager() back to pure in-memory (no store attached, no reconcile).

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}

Key	Meaning
lost	Orphaned RUNNING/PENDING jobs transitioned to LOST
redelivered	COMPLETED-but-undelivered jobs whose deliver-back was replayed
rehydrated	Total jobs re-loaded into the in-memory map

---

Best Practices

Call reconcile_on_start() exactly once, at startup

Run it once at boot, after all deliver-back handlers are wired — mirroring how OutboundQueue.drain_pending() is wired. Calling it later risks a partially-wired redeliver.

Make redeliver idempotent

redeliver receives the persisted JobInfo. A transient failure inside it means “retry on the next boot” — the job is left undelivered, never silently lost. Design the handler so a double-fire is harmless.

Sweep LOST records with cleanup_completed()

LOST is age-evictable by cleanup_completed(max_age=...). Run a periodic sweep so reconciled orphans don’t accumulate across restarts.

Keep the store thread-safe

upsert() is called from BackgroundJobManager’s worker threads. Guard shared state (or use a per-thread connection) so concurrent transitions don’t corrupt the store.

Opt out with the env var if you truly want in-memory

For the shared get_job_manager() the escape hatch is the environment variable, not the constructor kwarg — set PRAISONAI_BACKGROUND_JOB_STORE=0 to get pure in-memory behaviour (no store, no reconcile). When you construct BackgroundJobManager yourself, omitting store= keeps that manager in-memory with zero overhead.

---

Related

Background Tasks

The synchronous job manager and background runner patterns.

Background Subagents

Spawn subagents that deliver results back to chat when done.

Durable Delivery

Persist outbound bot messages with retry and crash-safe drain.

Hook Events

Subscribe to JOB_COMPLETED and other lifecycle events.