circuitforge-core/docs/modules/tasks.md

tasks

VRAM-aware background task scheduler. Manages a queue of LLM inference jobs and coordinates VRAM allocation with the cf-orch coordinator before executing each task.

from circuitforge_core.tasks import TaskScheduler, get_scheduler, reset_scheduler

Why VRAM-aware scheduling

Running multiple LLM inference jobs concurrently on a single GPU causes OOM errors and corrupted outputs. The scheduler serializes LLM work per service and negotiates with the cf-orch coordinator so tasks across multiple products don't compete for the same VRAM budget.

Core API

get_scheduler() -> TaskScheduler

Returns the singleton scheduler for the current process. Creates it on first call.

reset_scheduler()

Tears down the scheduler (releases VRAM leases, cancels pending tasks). Called during FastAPI lifespan teardown.

# In FastAPI lifespan
from circuitforge_core.tasks import get_scheduler, reset_scheduler

@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler = get_scheduler()
    yield
    reset_scheduler()

scheduler.submit(task_type, payload, vram_gb) -> str

Enqueues a task. Returns the task ID. The scheduler acquires a VRAM lease from the coordinator before executing.

task_id = await scheduler.submit(
    task_type="recipe_llm",
    payload={"pantry_ids": [1, 2, 3]},
    vram_gb=4.0,
)

scheduler.result(task_id) -> TaskResult | None

Polls for a completed result. Returns None if still running.

VRAM budgets

Each product defines its VRAM budgets in compose.yml / compose.override.yml:

environment:
  VRAM_BUDGET_RECIPE_LLM: "4.0"
  VRAM_BUDGET_EXPIRY_LLM: "2.0"

These map to task types in the scheduler. If the coordinator is unavailable (local dev without cf-orch), the scheduler falls back to sequential local execution.

Shim pattern

Products that need to re-export scheduler functions for backward compatibility use a shim:

# myproduct/app/tasks/scheduler.py
from circuitforge_core.tasks.scheduler import (
    get_scheduler as _base_get_scheduler,
    reset_scheduler,          # re-export for lifespan teardown
)

def get_scheduler():
    """Product-specific scheduler with service name injected."""
    return _base_get_scheduler(service_name="myproduct")

Always re-export reset_scheduler from the shim so the FastAPI lifespan can import it from one place.

Generic caller/args dispatch

Separate from the VRAM-budgeted scheduler above: dispatch_task()/get_task_status() are a generic, product-agnostic pair for running a named callable in the background and polling its status, keyed by a "product/task_name" string rather than the scheduler's task_id/job_id/SQLite-table shape.

from circuitforge_core.tasks import register_task_runner, dispatch_task, get_task_status

# Once, at product startup:
register_task_runner("pagepiper/ingest_pdf", run_ingest_pdf)

# Anywhere a task needs dispatching:
task_id = dispatch_task("pagepiper/ingest_pdf", {"doc_id": "...", "file_path": "..."})
get_task_status(task_id)  # {"status": "running", "progress": 0, "error": None}

Free-tier, in-process, single-node — runs on a background thread pool, no cross-node distribution. dispatch_task() raises LookupError for an unregistered caller, so products with an except Exception: ... fallback to local execution keep working unchanged if they forget to register a runner.

register_task_runner/dispatch_task/get_task_status route through circuitforge_core.tasks.dispatch internally — nothing shared with scheduler.py's VRAM-aware queue.