From 3f9677f6aae8d315acd5f3cdedb653456ce9275b Mon Sep 17 00:00:00 2001 From: Mahdi Bazrafshan Date: Sat, 22 Aug 2026 15:32:33 +0330 Subject: [PATCH] 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 --- docs/api-reference.md | 32 +++++++ docs/tasks.md | 2 +- src/benchmarking/benchmark_service.py | 20 ++++ src/benchmarking/routes.py | 83 +++++++++------- src/documents/routes.py | 37 +++++++- src/jobs/__init__.py | 1 + src/jobs/models.py | 48 ++++++++++ src/jobs/routes.py | 42 ++++++++ src/jobs/service.py | 114 ++++++++++++++++++++++ src/main.py | 2 + src/storage/sqlite.py | 132 +++++++++++++++++++++++++- 11 files changed, 473 insertions(+), 40 deletions(-) create mode 100644 src/jobs/__init__.py create mode 100644 src/jobs/models.py create mode 100644 src/jobs/routes.py create mode 100644 src/jobs/service.py diff --git a/docs/api-reference.md b/docs/api-reference.md index a0265c6..f7df808 100644 --- a/docs/api-reference.md +++ b/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: diff --git a/docs/tasks.md b/docs/tasks.md index 6c794de..279c0ff 100644 --- a/docs/tasks.md +++ b/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 diff --git a/src/benchmarking/benchmark_service.py b/src/benchmarking/benchmark_service.py index 920dd8e..1f81967 100644 --- a/src/benchmarking/benchmark_service.py +++ b/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. diff --git a/src/benchmarking/routes.py b/src/benchmarking/routes.py index 87df44a..a044a0c 100644 --- a/src/benchmarking/routes.py +++ b/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) diff --git a/src/documents/routes.py b/src/documents/routes.py index d1f6bb4..f106763 100644 --- a/src/documents/routes.py +++ b/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) diff --git a/src/jobs/__init__.py b/src/jobs/__init__.py new file mode 100644 index 0000000..8780153 --- /dev/null +++ b/src/jobs/__init__.py @@ -0,0 +1 @@ +"""Background job tracking for long-running benchmark and process operations.""" diff --git a/src/jobs/models.py b/src/jobs/models.py new file mode 100644 index 0000000..d6b6a03 --- /dev/null +++ b/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 diff --git a/src/jobs/routes.py b/src/jobs/routes.py new file mode 100644 index 0000000..192a210 --- /dev/null +++ b/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) diff --git a/src/jobs/service.py b/src/jobs/service.py new file mode 100644 index 0000000..8194423 --- /dev/null +++ b/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"}, + ) diff --git a/src/main.py b/src/main.py index 6f1d89b..b539a4b 100644 --- a/src/main.py +++ b/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") diff --git a/src/storage/sqlite.py b/src/storage/sqlite.py index 1216447..c3c4a13 100644 --- a/src/storage/sqlite.py +++ b/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]: