Background Task

Background work — Prefect, and one in-process handoff

Read docs/howto/add-background-task.md. Flows live in backend/app/worker/tasks/ (rag_tasks.py, mcp_tasks.py, report_tasks.py); deployments are registered in app/worker/prefect_app.py.

Pick the right one

from app.core.background import spawn, spawn_after_commit

spawn(send_the_slow_thing(user_id), name="notify-owner")

# Reads a row the caller has just written, so it waits for the commit.
spawn_after_commit(self.db, ingest_document_flow(rag_document_id=str(doc.id)), name=...)

A flow that reads its own row by id takes spawn_after_commit. spawn creates the task at once and the loop starts it at the next suspension point, which inside a request is well before the commit — so the flow opens its own session, cannot see the uncommitted row, and stops. That is #417: an upload answered {"status": "processing"} that stayed that way. spawn_after_commit queues the coroutine on the session and the session starts it two statements after commit(); a session that rolls back closes it instead, with a warning. Works from a route, a WebSocket handler, the CLI or a worker, because the queue belongs to the session rather than to a response.

drain(timeout=30.0) is the shutdown counterpart — in-flight tasks get a chance to finish rather than being cut off mid-write. A CLI command that dispatches work must call it before returning: asyncio.run cancels whatever is still pending, so without it the command kills the sync it just reported starting.

Never bare asyncio.create_task. The event loop holds it only weakly: drop the reference and the task can be garbage-collected mid-flight — the classic symptom is an ingestion that works under load and vanishes when the system is idle, with nothing in the logs. Worse, an exception inside a discarded task is never retrieved, so a flow that raises fails completely silently.

app/core/background.py fixes both once: it holds a strong reference until the task finishes and attaches a done-callback that surfaces whatever went wrong. It is in the gated platform layer at 100%. Use it, or use Prefect.

app/worker/background/ (channel, rag) is the older FastAPI-BackgroundTasks fallback, dispatched with its own fire_and_forget(coro, *, label=...): in-process, no retry, no persistence, and CPU-bound work there starves the event loop. Prefer spawn for new code and Prefect for anything that must survive a restart.

Add a flow

from prefect import flow

@flow(name="ingest-document", log_prints=True)
async def ingest_document_flow(document_id: str) -> dict[str, Any]: ...

1. Define it in app/worker/tasks/<area>.py.
2. Dispatch from a service, not a route. Business logic stays in services/; enqueue at the end of the unit of work.
3. Schedule it, if periodic, with a CronSchedule/IntervalSchedule deployment in app/worker/prefect_app.py.
4. Verify: prefect-server and prefect-runner start with make dev; runs are at http://localhost:4200. PREFECT_API_URL points at the self-hosted server, or a Cloud workspace plus PREFECT_API_KEY.

Rules

• Serializable args only — ids and primitives, never ORM objects or a session. Re-fetch inside the flow with a fresh session.
• Idempotent where possible. A retry must not double-charge, double-index or double-send.
• Heavy imports inside the function, to keep the API import-light.
• Record what was spent even when the flow fails. Embedding spend goes to ingestion_spend (which carries cost_is_partial for models the price snapshot does not know). A cost that only lands on success is a budget with a hole in it.
• A flow that touches an org-scoped resource still needs the organization id — there is no ambient tenant in a worker.