feat(jobs): add background job support for benchmarks and processing

Why:
- Long benchmark and process runs block HTTP clients; task #20 required async execution.

Changes:
- SQLite jobs table + GET /jobs and GET /jobs/{id}
- POST /benchmarks?background=true and POST /documents/{id}/process?background=true return 202 + job_id
- FastAPI BackgroundTasks execute work in-process; sync paths unchanged

Impact:
- Jobs are lost on server restart (Option A, no external queue)

Co-authored-by: Cursor <cursoragent@cursor.com>

docs/api-reference.md

@@ -82,6 +82,9 @@ Delete a document and its vectors.
 
 Run chunking strategies on a document.
 
+**Query parameters:**
+- `background` (bool, default `false`) — when `true`, enqueue processing and return **202** with a `job_id`; poll `GET /jobs/{job_id}`.
+
 **Request:**
 ```json
 {
@@ -199,6 +202,9 @@ Retrieve a past query.
 
 Run a benchmark comparing multiple strategies.
 
+**Query parameters:**
+- `background` (bool, default `false`) — when `true`, enqueue the run and return **202** with a `job_id`; poll `GET /jobs/{job_id}` for status and result.
+
 **Request:**
 ```json
 {
@@ -278,6 +284,32 @@ List all experiments.
 
 ---
 
+### Jobs
+
+Background execution for long-running benchmarks and document processing (FastAPI `BackgroundTasks` + SQLite job records).
+
+#### `GET /jobs`
+
+List jobs (newest first).
+
+**Query parameters:** `job_type`, `status`, `offset`, `limit`
+
+#### `GET /jobs/{job_id}`
+
+Poll job status. When `status` is `completed`, `result` contains the same payload as the synchronous endpoint would return; when `failed`, `error` is set.
+
+**Response (202 enqueue body from POST with `background=true`):**
+```json
+{
+  "job_id": "abc123",
+  "job_type": "benchmark",
+  "status": "pending",
+  "poll_url": "/jobs/abc123"
+}
+```
+
+---
+
 ## Error Responses
 
 All errors return:

docs/tasks.md

@@ -43,7 +43,7 @@ Status legend: `DONE` `IN_PROGRESS` `TODO`
 | 17 | Create question-answer evaluation dataset from insurance regulation document | DONE | 12 |
 | 18 | Implement HTML benchmark report generation system | DONE | 14 |
 | 19 | Design HTML report structure for experiment comparison and visualization | DONE | 14 |
-| 20 | Create background processing jobs for document ingestion and benchmarking | TODO | 12 |
+| 20 | Create background processing jobs for document ingestion and benchmarking | DONE | 12 |
 
 ## Phase 5 — Wiring + Verification
 

src/benchmarking/benchmark_service.py

@@ -57,6 +57,26 @@ def load_questions(file_path: str | Path) -> list[dict]:
         raise BenchmarkError(f"Invalid JSON in questions file: {exc}") from exc
 
 
+def resolve_benchmark_questions(
+    *,
+    questions: list[dict] | None = None,
+    questions_file: str | None = None,
+) -> list[dict]:
+    """Resolve and validate benchmark questions from inline list or file."""
+    if questions_file:
+        return load_questions(questions_file)
+    if questions and len(questions) > 0:
+        valid_questions = [
+            q for q in questions if isinstance(q, dict) and "question" in q
+        ]
+        if not valid_questions:
+            raise BenchmarkError(
+                "Invalid questions: each question must have a 'question' field"
+            )
+        return valid_questions
+    raise BenchmarkError("Either 'questions' or 'questions_file' must be provided")
+
+
 def load_questions_from_string(questions_json: str) -> list[dict]:
     """Load questions from a JSON string.
 

src/benchmarking/routes.py

@@ -3,17 +3,19 @@
 Endpoints:
   POST   /queries              Ask a question against a strategy
   GET    /queries/{id}         Retrieve a past query
-  POST   /benchmarks           Run a benchmark (or dry run)
+  POST   /benchmarks           Run a benchmark (or dry run); ?background=true enqueues a job
   GET    /benchmarks/{id}      Retrieve experiment results
   GET    /experiments          List all experiments
 """
 
-from fastapi import APIRouter, Query
-from fastapi.responses import HTMLResponse
+from fastapi import APIRouter, BackgroundTasks, Query
+from fastapi.responses import HTMLResponse, JSONResponse
 
 from src.core.exceptions import BenchmarkError, QueryError
 from src.core.models import StrategyName
 from src.benchmarking import benchmark_service, query_service
+from src.jobs import service as jobs_service
+from src.jobs.models import JobCreatedResponse
 from src.benchmarking.models import (
     BenchmarkRequest,
     BenchmarkResponse,
@@ -88,50 +90,64 @@ async def get_query(query_id: str):
 
 # ── Benchmark Endpoints ──────────────────────────────────────────
 
-@router.post("/benchmarks", response_model=BenchmarkResponse, status_code=201)
-async def create_benchmark(request: BenchmarkRequest):
+@router.post("/benchmarks")
+async def create_benchmark(
+    request: BenchmarkRequest,
+    background_tasks: BackgroundTasks,
+    background: bool = Query(
+        False, description="Run in background; returns job_id (HTTP 202)"
+    ),
+):
     """Run a benchmark comparing multiple strategies on multiple questions.
 
     Can run in dry_run mode to get cost estimate without executing.
+    Pass ``background=true`` to enqueue and poll ``GET /jobs/{job_id}``.
     """
-    # Load questions - prefer questions_file over inline questions
-    if request.questions_file:
-        questions = benchmark_service.load_questions(request.questions_file)
-    elif request.questions and len(request.questions) > 0:
-        # Validate that questions have required fields
-        valid_questions = [
-            q for q in request.questions
-            if isinstance(q, dict) and "question" in q
-        ]
-        if not valid_questions:
-            raise BenchmarkError("Invalid questions: each question must have a 'question' field")
-        questions = valid_questions
-    else:
-        raise BenchmarkError("Either 'questions' or 'questions_file' must be provided")
+    questions = benchmark_service.resolve_benchmark_questions(
+        questions=request.questions,
+        questions_file=request.questions_file,
+    )
 
     if not questions:
         raise BenchmarkError("No questions to benchmark")
 
-    # Dry run - return cost estimate only
     if request.dry_run:
+        if background:
+            raise BenchmarkError("dry_run cannot be used with background=true")
         estimate = benchmark_service.estimate_cost(
             num_questions=len(questions),
             num_strategies=len(request.strategies),
         )
-        # Return as BenchmarkResponse with minimal data
-        return BenchmarkResponse(
-            experiment_id="dry_run",
-            document_id=request.document_id,
-            strategies_used=[s.value for s in request.strategies],
-            questions_count=len(questions),
-            aggregate_metrics={},
-            best_strategy="N/A (dry run)",
-            total_latency_seconds=0,
-            estimated_cost_usd=estimate["estimated_cost_usd"],
-            created_at="N/A",
+        return JSONResponse(
+            status_code=201,
+            content=BenchmarkResponse(
+                experiment_id="dry_run",
+                document_id=request.document_id,
+                strategies_used=[s.value for s in request.strategies],
+                questions_count=len(questions),
+                aggregate_metrics={},
+                best_strategy="N/A (dry run)",
+                total_latency_seconds=0,
+                estimated_cost_usd=estimate["estimated_cost_usd"],
+                created_at="N/A",
+            ).model_dump(),
         )
 
-    # Run full benchmark
+    if background:
+        payload = request.model_dump(mode="json")
+        payload["questions"] = questions
+        payload.pop("questions_file", None)
+        payload.pop("dry_run", None)
+        job = jobs_service.enqueue_benchmark(payload)
+        background_tasks.add_task(jobs_service.run_benchmark_job, job["id"])
+        body = JobCreatedResponse(
+            job_id=job["id"],
+            job_type=job["job_type"],
+            status=job["status"],
+            poll_url=f"/jobs/{job['id']}",
+        )
+        return JSONResponse(status_code=202, content=body.model_dump())
+
     result = benchmark_service.run_benchmark(
         document_id=request.document_id,
         strategies=request.strategies,
@@ -142,7 +158,7 @@ async def create_benchmark(request: BenchmarkRequest):
         corpus_model_id=request.corpus_model_id,
     )
 
-    return BenchmarkResponse(
+    response = BenchmarkResponse(
         experiment_id=result["experiment_id"],
         document_id=result["document_id"],
         strategies_used=result["strategies_used"],
@@ -155,6 +171,7 @@ async def create_benchmark(request: BenchmarkRequest):
         estimated_cost_usd=result["estimated_cost_usd"],
         created_at=result["created_at"],
     )
+    return JSONResponse(status_code=201, content=response.model_dump())
 
 
 @router.get("/benchmarks/{experiment_id}", response_model=ExperimentDetailResponse)

src/documents/routes.py

@@ -2,12 +2,13 @@
 
 Endpoints:
   POST   /documents              Upload a .docx / .doc / .pdf file
-  POST   /documents/{id}/process  Run chunking strategies
+  POST   /documents/{id}/process  Run chunking strategies; ?background=true enqueues a job
   DELETE /documents/{id}          Remove document + vectors
   GET    /strategies              List available chunking strategies
 """
 
-from fastapi import APIRouter, File, UploadFile
+from fastapi import APIRouter, BackgroundTasks, File, Query, UploadFile
+from fastapi.responses import JSONResponse
 
 from src.core.exceptions import DocumentProcessingError
 from src.documents.models import (
@@ -21,6 +22,9 @@ from src.documents.models import (
 )
 from src.documents.parser import SUPPORTED_SUFFIXES
 from src.documents import service
+from src.jobs import service as jobs_service
+from src.jobs.models import JobCreatedResponse
+from src.storage import sqlite as db
 
 router = APIRouter()
 
@@ -63,9 +67,32 @@ async def upload_document(file: UploadFile = File(...)):
     )
 
 
-@router.post("/documents/{doc_id}/process", response_model=ProcessResponse)
-async def process_document(doc_id: str, request: ProcessRequest):
-    """Run selected chunking strategies on an uploaded document."""
+@router.post("/documents/{doc_id}/process")
+async def process_document(
+    doc_id: str,
+    request: ProcessRequest,
+    background_tasks: BackgroundTasks,
+    background: bool = Query(
+        False, description="Run in background; returns job_id (HTTP 202)"
+    ),
+):
+    """Run selected chunking strategies on an uploaded document.
+
+    Pass ``background=true`` to enqueue and poll ``GET /jobs/{job_id}``.
+    """
+    if background:
+        if db.get_document(doc_id) is None:
+            raise DocumentProcessingError(f"Document not found: {doc_id}")
+        job = jobs_service.enqueue_process(doc_id, request)
+        background_tasks.add_task(jobs_service.run_process_job, job["id"])
+        body = JobCreatedResponse(
+            job_id=job["id"],
+            job_type=job["job_type"],
+            status=job["status"],
+            poll_url=f"/jobs/{job['id']}",
+        )
+        return JSONResponse(status_code=202, content=body.model_dump())
+
     return service.process_document(doc_id, request)
 
 

src/jobs/__init__.py

@@ -0,0 +1 @@
+"""Background job tracking for long-running benchmark and process operations."""

src/jobs/models.py

@@ -0,0 +1,48 @@
+"""Request/response schemas for the Jobs API."""
+
+from enum import Enum
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+
+class JobType(str, Enum):
+    BENCHMARK = "benchmark"
+    PROCESS = "process"
+
+
+class JobStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class JobCreatedResponse(BaseModel):
+    """Returned when a long-running operation is enqueued (HTTP 202)."""
+    job_id: str
+    job_type: str
+    status: str = "pending"
+    poll_url: str
+
+
+class JobDetailResponse(BaseModel):
+    """Full job status including result or error when finished."""
+    id: str
+    job_type: str
+    status: str
+    payload: dict[str, Any] = Field(default_factory=dict)
+    result: Optional[dict[str, Any]] = None
+    error: Optional[str] = None
+    progress: dict[str, Any] = Field(default_factory=dict)
+    created_at: str
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+
+
+class JobListResponse(BaseModel):
+    """Paginated list of jobs."""
+    items: list[JobDetailResponse]
+    total: int
+    offset: int
+    limit: int

src/jobs/routes.py

@@ -0,0 +1,42 @@
+"""Job status API routes.
+
+Endpoints:
+  GET  /jobs           List background jobs
+  GET  /jobs/{id}      Poll job status and result
+"""
+
+from fastapi import APIRouter, Query
+
+from src.core.exceptions import BenchmarkError
+from src.jobs.models import JobDetailResponse, JobListResponse
+from src.storage import sqlite as db
+
+router = APIRouter(tags=["jobs"])
+
+
+@router.get("/jobs", response_model=JobListResponse)
+async def list_jobs(
+    job_type: str | None = Query(None, description="Filter by job_type (benchmark|process)"),
+    status: str | None = Query(None, description="Filter by status"),
+    offset: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=200),
+):
+    """List background jobs, newest first."""
+    result = db.list_jobs(
+        job_type=job_type, status=status, offset=offset, limit=limit
+    )
+    return JobListResponse(
+        items=[JobDetailResponse(**item) for item in result["items"]],
+        total=result["total"],
+        offset=result["offset"],
+        limit=result["limit"],
+    )
+
+
+@router.get("/jobs/{job_id}", response_model=JobDetailResponse)
+async def get_job(job_id: str):
+    """Poll a background job by ID."""
+    job = db.get_job(job_id)
+    if job is None:
+        raise BenchmarkError(f"Job not found: {job_id}")
+    return JobDetailResponse(**job)

src/jobs/service.py

@@ -0,0 +1,114 @@
+"""Background job enqueue and execution via FastAPI BackgroundTasks."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from src.benchmarking import benchmark_service
+from src.core.models import StrategyName
+from src.documents import service as documents_service
+from src.documents.models import ProcessRequest
+from src.storage import sqlite as db
+
+logger = logging.getLogger(__name__)
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def enqueue_benchmark(payload: dict[str, Any]) -> dict[str, Any]:
+    """Create a pending benchmark job with a resolved questions payload."""
+    return db.create_job(job_type="benchmark", payload=payload)
+
+
+def enqueue_process(document_id: str, request: ProcessRequest) -> dict[str, Any]:
+    """Create a pending document processing job."""
+    payload = {"document_id": document_id, **request.model_dump(mode="json")}
+    return db.create_job(job_type="process", payload=payload)
+
+
+def run_benchmark_job(job_id: str) -> None:
+    """Execute a benchmark job (runs in a FastAPI background task)."""
+    job = db.get_job(job_id)
+    if job is None:
+        logger.error("Benchmark job not found: %s", job_id)
+        return
+
+    payload = job["payload"]
+    try:
+        db.update_job(
+            job_id,
+            status="running",
+            started_at=_now(),
+            progress={"message": "Running benchmark"},
+        )
+        strategies = [StrategyName(s) for s in payload["strategies"]]
+        result = benchmark_service.run_benchmark(
+            document_id=payload["document_id"],
+            strategies=strategies,
+            questions=payload["questions"],
+            top_k=payload.get("top_k", 5),
+            neighbor_prev=payload.get("neighbor_prev", 0),
+            neighbor_next=payload.get("neighbor_next", 0),
+            corpus_model_id=payload.get("corpus_model_id"),
+        )
+        db.update_job(
+            job_id,
+            status="completed",
+            result=result,
+            completed_at=_now(),
+            progress={"message": "Benchmark completed"},
+        )
+    except Exception as exc:
+        logger.exception("Benchmark job %s failed", job_id)
+        db.update_job(
+            job_id,
+            status="failed",
+            error=str(exc),
+            completed_at=_now(),
+            progress={"message": "Benchmark failed"},
+        )
+
+
+def run_process_job(job_id: str) -> None:
+    """Execute a document processing job (runs in a FastAPI background task)."""
+    job = db.get_job(job_id)
+    if job is None:
+        logger.error("Process job not found: %s", job_id)
+        return
+
+    payload = job["payload"]
+    document_id = payload["document_id"]
+    request = ProcessRequest(
+        strategies=[StrategyName(s) for s in payload["strategies"]],
+        boundary_model_id=payload.get("boundary_model_id"),
+        corpus_model_id=payload.get("corpus_model_id"),
+    )
+
+    try:
+        db.update_job(
+            job_id,
+            status="running",
+            started_at=_now(),
+            progress={"message": "Processing document"},
+        )
+        response = documents_service.process_document(document_id, request)
+        db.update_job(
+            job_id,
+            status="completed",
+            result=response.model_dump(mode="json"),
+            completed_at=_now(),
+            progress={"message": "Processing completed"},
+        )
+    except Exception as exc:
+        logger.exception("Process job %s failed", job_id)
+        db.update_job(
+            job_id,
+            status="failed",
+            error=str(exc),
+            completed_at=_now(),
+            progress={"message": "Processing failed"},
+        )

src/main.py

@@ -21,6 +21,7 @@ from src.core.exceptions import (
 from src.documents.routes import router as documents_router
 from src.benchmarking.routes import router as benchmarking_router
 from src.admin.routes import router as admin_router
+from src.jobs.routes import router as jobs_router
 from src.storage.sqlite import init_db
 
 # Configure logging
@@ -95,6 +96,7 @@ def create_app() -> FastAPI:
     app.include_router(documents_router)
     app.include_router(benchmarking_router)
     app.include_router(admin_router)
+    app.include_router(jobs_router)
 
     # Mount dashboard at /app
     static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static")

src/storage/sqlite.py

@@ -88,6 +88,19 @@ CREATE TABLE IF NOT EXISTS app_settings (
     key   TEXT PRIMARY KEY,
     value TEXT NOT NULL
 );
+
+CREATE TABLE IF NOT EXISTS jobs (
+    id           TEXT PRIMARY KEY,
+    job_type     TEXT NOT NULL,
+    status       TEXT NOT NULL,
+    payload      TEXT NOT NULL DEFAULT '{}',
+    result       TEXT,
+    error        TEXT,
+    progress     TEXT NOT NULL DEFAULT '{}',
+    created_at   TEXT NOT NULL,
+    started_at   TEXT,
+    completed_at TEXT
+);
 """
 
 
@@ -149,6 +162,21 @@ def _migrate_schema(conn: sqlite3.Connection) -> None:
         (LEGACY_CLOUD_MODEL_ID, Provider.CLOUD.value),
     )
 
+    conn.execute(
+        """CREATE TABLE IF NOT EXISTS jobs (
+            id           TEXT PRIMARY KEY,
+            job_type     TEXT NOT NULL,
+            status       TEXT NOT NULL,
+            payload      TEXT NOT NULL DEFAULT '{}',
+            result       TEXT,
+            error        TEXT,
+            progress     TEXT NOT NULL DEFAULT '{}',
+            created_at   TEXT NOT NULL,
+            started_at   TEXT,
+            completed_at TEXT
+        )"""
+    )
+
 
 def get_app_setting(key: str) -> str | None:
     """Read a persisted app setting value."""
@@ -445,12 +473,114 @@ def delete_experiment(experiment_id: str) -> bool:
         conn.close()
 
 
+# ── Job CRUD ───────────────────────────────────────────────────────
+
+def create_job(*, job_type: str, payload: dict[str, Any]) -> dict[str, Any]:
+    """Insert a pending background job."""
+    job_id = _new_id()
+    conn = _connect()
+    try:
+        conn.execute(
+            """INSERT INTO jobs
+               (id, job_type, status, payload, progress, created_at)
+               VALUES (?, ?, ?, ?, ?, ?)""",
+            (
+                job_id,
+                job_type,
+                "pending",
+                json.dumps(payload),
+                json.dumps({}),
+                _now(),
+            ),
+        )
+        conn.commit()
+        return get_job(job_id)  # type: ignore[return-value]
+    finally:
+        conn.close()
+
+
+def get_job(job_id: str) -> dict[str, Any] | None:
+    """Fetch a job by ID."""
+    conn = _connect()
+    try:
+        row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone()
+        if row is None:
+            return None
+        return _row_to_dict(row)
+    finally:
+        conn.close()
+
+
+def update_job(job_id: str, **fields: Any) -> None:
+    """Update job fields (status, result, error, progress, timestamps)."""
+    if not fields:
+        return
+
+    json_fields = {"payload", "result", "progress"}
+    sets: list[str] = []
+    values: list[Any] = []
+    for key, value in fields.items():
+        if key in json_fields and value is not None and not isinstance(value, str):
+            value = json.dumps(value)
+        sets.append(f"{key} = ?")
+        values.append(value)
+
+    values.append(job_id)
+    conn = _connect()
+    try:
+        conn.execute(
+            f"UPDATE jobs SET {', '.join(sets)} WHERE id = ?",
+            values,
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def list_jobs(
+    *,
+    job_type: str | None = None,
+    status: str | None = None,
+    offset: int = 0,
+    limit: int = 50,
+) -> dict[str, Any]:
+    """List jobs with optional filters, newest first."""
+    limit = max(1, min(int(limit), 200))
+    offset = max(0, int(offset))
+    conn = _connect()
+    try:
+        clauses: list[str] = []
+        params: list[Any] = []
+        if job_type:
+            clauses.append("job_type = ?")
+            params.append(job_type)
+        if status:
+            clauses.append("status = ?")
+            params.append(status)
+        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
+        total = conn.execute(
+            f"SELECT COUNT(*) FROM jobs {where}", params
+        ).fetchone()[0]
+        rows = conn.execute(
+            f"SELECT * FROM jobs {where} ORDER BY created_at DESC LIMIT ? OFFSET ?",
+            [*params, limit, offset],
+        ).fetchall()
+        return {
+            "items": [_row_to_dict(r) for r in rows],
+            "total": total,
+            "offset": offset,
+            "limit": limit,
+        }
+    finally:
+        conn.close()
+
+
 # ── Internal helpers ───────────────────────────────────────────────
 
 _JSON_FIELDS = {"chunk_counts", "retrieved_chunks", "expansion_tree", "latency_breakdown",
                 "token_usage", "document_tree", "benchmark_config",
                 "questions", "per_question", "aggregate_metrics",
-                "strategies_used"}
+                "strategies_used", "payload", "result", "progress"}
 
 
 def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]: