Extraction layer: start and poll an extraction

.github/workflows/lint.yml

@@ -4,7 +4,7 @@ on:
   push:
     branches: [main,development]
   pull_request:
-    branches: [main,development]
+    branches: [main, development, 'development-*']
 
 jobs:
   lint:
@@ -23,7 +23,7 @@ jobs:
         uses: astral-sh/setup-uv@v6
 
       - name: Install linter
-        run: uv pip install --system ruff
+        run: uv pip install --system ruff==0.16.1
 
       - name: Run linter
         run: |

.github/workflows/tests.yml

@@ -9,7 +9,7 @@ on:
       - 'alembic.ini'
       - '.github/workflows/tests.yml'
   pull_request:
-    branches: [main,development]
+    branches: [main, development, 'development-*']
     paths:
       - '**.py'
       - 'requirements.txt'

app/api/router.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter
 
-from app.api.routes import forms, input, jobs, system, templates, weather, zipcode
+from app.api.routes import extraction, forms, input, jobs, system, templates, weather, zipcode
 from app.core.config import API_PREFIX
 
 api_router = APIRouter()
@@ -10,4 +10,5 @@
 api_router.include_router(jobs.router, prefix=API_PREFIX)
 api_router.include_router(weather.router, prefix=API_PREFIX)
 api_router.include_router(zipcode.router, prefix=API_PREFIX)
-api_router.include_router(input.router, prefix=API_PREFIX)
+api_router.include_router(input.router, prefix=API_PREFIX)
+api_router.include_router(extraction.router, prefix=API_PREFIX)

app/api/routes/extraction.py

@@ -0,0 +1,131 @@
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+from sqlmodel import Session
+
+from app.api.deps import get_db
+from app.api.schemas.enums import ExtractionStatus, InputStatus
+from app.api.schemas.extraction import (
+    ExtractionCompleted,
+    ExtractionJobResponse,
+    ExtractionProcessing,
+    ExtractionRequest,
+)
+from app.api.schemas.incident_contract import IncidentContract
+from app.core.config import (
+    ESTIMATED_EXTRACTION_SECONDS,
+    EXTRACTION_POLL_INTERVAL_SECONDS,
+)
+from app.core.errors.base import AppError
+from app.db.repositories import (
+    get_extraction,
+    get_extraction_by_input,
+    get_incident_by_extract,
+    get_input,
+)
+from app.services.extraction import ExtractionService
+from app.services.llm import check_ollama_available
+
+router = APIRouter(prefix="/extract", tags=["extraction"])
+
+
+@router.post("/{input_id}", response_model=ExtractionJobResponse, status_code=202)
+def create_extraction(
+    input_id: UUID,
+    body: ExtractionRequest | None = None,
+    db: Session = Depends(get_db),
+):
+    record = get_input(db, input_id)
+    if record is None:
+        raise AppError(
+            f"Input with ID {input_id} not found",
+            status_code=404,
+            error_code="INPUT_NOT_FOUND",
+        )
+
+    if record.status != InputStatus.ready:
+        raise AppError(
+            f"Input is in '{record.status}' state. Wait until status is 'ready'.",
+            status_code=409,
+            error_code="INPUT_NOT_READY",
+            detail={"current_status": record.status},
+        )
+
+    existing = get_extraction_by_input(db, input_id)
+    if existing is not None:
+        raise AppError(
+            "An extraction already exists for this input",
+            status_code=409,
+            error_code="EXTRACTION_EXISTS",
+            detail={"existing_extract_id": str(existing.extract_id)},
+        )
+
+    if not check_ollama_available():
+        raise AppError(
+            "Ollama LLM service is not available",
+            status_code=503,
+            error_code="LLM_UNAVAILABLE",
+        )
+
+    model_override = body.model_override if body else None
+    svc = ExtractionService()
+    extraction, job = svc.start_extraction(db, input_id, model_override=model_override)
+
+    return ExtractionJobResponse(
+        extract_id=extraction.extract_id,
+        input_id=input_id,
+        job_id=job.job_id,
+        status=extraction.status,
+        queued_at=extraction.created_at,
+        estimated_seconds=ESTIMATED_EXTRACTION_SECONDS,
+        poll_url=f"/api/v1/extract/{extraction.extract_id}",
+    )
+
+
+@router.get("/{extract_id}", response_model=ExtractionCompleted | ExtractionProcessing)
+def get_extraction_result(extract_id: UUID, db: Session = Depends(get_db)):
+    extraction = get_extraction(db, extract_id)
+    if extraction is None:
+        raise AppError(
+            f"Extraction with ID {extract_id} not found",
+            status_code=404,
+            error_code="EXTRACT_NOT_FOUND",
+        )
+
+    if extraction.status == ExtractionStatus.completed:
+        incident = get_incident_by_extract(db, extract_id)
+        contract = IncidentContract.model_validate(
+            (incident.incident_contract if incident else None) or {}
+        )
+        return ExtractionCompleted(
+            extract_id=extraction.extract_id,
+            input_id=extraction.input_id,
+            incident_id=incident.incident_id if incident else None,
+            status="completed",
+            incident_contract=contract,
+            completed_at=extraction.completed_at,
+            model_used=extraction.model_used,
+            processing_time_seconds=extraction.processing_time_seconds,
+            corrections=extraction.corrections,
+        )
+
+    retry_after = (
+        EXTRACTION_POLL_INTERVAL_SECONDS
+        if extraction.status == ExtractionStatus.processing
+        else None
+    )
+    partial = (
+        IncidentContract.model_validate(extraction.partial_result)
+        if extraction.partial_result
+        else None
+    )
+    return ExtractionProcessing(
+        extract_id=extraction.extract_id,
+        input_id=extraction.input_id,
+        status=extraction.status,
+        started_at=extraction.started_at,
+        retry_after_seconds=retry_after,
+        error_type=extraction.error_type,
+        error_detail=extraction.error_detail,
+        partial_result=partial,
+    )

app/api/schemas/extraction.py

@@ -4,7 +4,7 @@
 from typing import Any, Literal
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict
 
 from app.api.schemas.enums import FieldSource
 from app.api.schemas.incident_contract import IncidentContract
@@ -47,6 +47,20 @@ class ExtractionRequest(BaseModel):
 # Responses
 # ---------------------------------------------------------------------------
 
+class ExtractionJobResponse(BaseModel):
+    """202 body for POST /extract/{input_id}. Carries the ids the client needs
+    to poll the extraction plus the underlying async job."""
+
+    extract_id: UUID
+    input_id: UUID
+    job_id: str
+    job_type: str = "extraction"
+    status: str
+    queued_at: datetime | None = None
+    estimated_seconds: int | None = None
+    poll_url: str
+
+
 class Correction(BaseModel):
     """One manual correction applied to the contract via PATCH."""
 

app/core/celery.py

@@ -16,7 +16,7 @@
     result_expires=86400,
 )
 
-celery_app.conf.include = ["app.tasks.fill", "app.tasks.purge", "app.tasks.transcribe"]
+celery_app.conf.include = ["app.tasks.fill", "app.tasks.purge", "app.tasks.transcribe", "app.tasks.extract"]
 
 # Optional Celery Beat schedule — runs purge_old_submissions once a day.
 # Enable by running: celery -A app.core.celery beat

app/core/config.py

@@ -58,6 +58,13 @@
 # or transcribing. Value matches the contract example (contracts/path/input.yaml).
 INPUT_POLL_INTERVAL_SECONDS = 5
 
+# Polling hint returned by GET /extract/{id} while an extraction is still
+# processing. Matches the contract example (contracts/path/extraction.yaml).
+EXTRACTION_POLL_INTERVAL_SECONDS = 5
+
+# Advisory estimate returned in the 202 body of POST /extract/{input_id}.
+ESTIMATED_EXTRACTION_SECONDS = int(os.getenv("ESTIMATED_EXTRACTION_SECONDS", "60"))
+
 # --- API Versioning -------------------------------------------------------
 API_PREFIX = "/api/v1"
 

app/core/errors/handlers.py

@@ -1,3 +1,5 @@
+import json
+
 from fastapi import Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
@@ -6,6 +8,23 @@
 from app.core.errors.base import AppError
 
 
+def _jsonable(value):
+    """Return value untouched if JSON-serializable, else a string form.
+
+    FastAPI puts the offending input on each validation error. When a JSON-body
+    endpoint is called with the wrong content-type, that input is the raw
+    request body as bytes, which JSONResponse cannot encode. Coerce anything
+    non-serializable so the 422 renders instead of turning into a 500.
+    """
+    try:
+        json.dumps(value)
+        return value
+    except (TypeError, ValueError):
+        if isinstance(value, bytes):
+            return value.decode("utf-8", "replace")
+        return str(value)
+
+
 def register_exception_handlers(app):
     @app.exception_handler(AppError)
     async def app_error_handler(request: Request, exc: AppError):
@@ -25,7 +44,7 @@ async def validation_error_handler(request: Request, exc: RequestValidationError
             validation_errors.append({
                 "field": field or None,
                 "issue": error.get("msg"),
-                "value": error.get("input"),
+                "value": _jsonable(error.get("input")),
             })
         return JSONResponse(
             status_code=422,

app/db/repositories.py

@@ -102,6 +102,11 @@ def get_extraction(session: Session, extract_id: UUID) -> Extraction | None:
     return session.get(Extraction, extract_id)
 
 
+def get_extraction_by_input(session: Session, input_id: UUID) -> Extraction | None:
+    statement = select(Extraction).where(Extraction.input_id == input_id)
+    return session.exec(statement).first()
+
+
 def update_extraction(session: Session, extraction: Extraction) -> Extraction:
     session.add(extraction)
     session.commit()

app/services/extraction.py

@@ -0,0 +1,53 @@
+"""Extraction service.
+
+Owns the write path that turns a ready input into a queued extraction: it
+creates the extraction row, creates the async job, and dispatches the worker.
+The route stays a thin HTTP handler and calls straight into here.
+
+The worker itself is stubbed for now (app/tasks/extract.py); the real chunked
+extraction is #630.
+"""
+
+from datetime import datetime, timezone
+
+from sqlmodel import Session
+
+from app.api.schemas.enums import ExtractionStatus
+from app.db.repositories import create_extraction, create_job, update_job
+from app.models import Extraction, Job
+from app.tasks.extract import extract_task
+
+
+class ExtractionService:
+    def start_extraction(
+        self,
+        session: Session,
+        input_id,
+        model_override: str | None = None,
+    ) -> tuple[Extraction, Job]:
+        """Create the extraction row and job, then dispatch the worker.
+
+        The extraction starts in ``processing`` so a poll right after the 202
+        sees the in-flight shape. Mirrors the transcription flow: the job row is
+        created first with a known job_id, dispatched, then its celery_task_id
+        is backfilled once the broker returns a task id.
+        """
+        now = datetime.now(timezone.utc)
+        extraction = Extraction(
+            input_id=input_id,
+            status=ExtractionStatus.processing,
+            started_at=now,
+            model_used=model_override,
+            created_at=now,
+            updated_at=now,
+        )
+        extraction = create_extraction(session, extraction)
+
+        job = Job(celery_task_id="", job_type="extraction", status="queued", model=model_override)
+        job = create_job(session, job)
+
+        result = extract_task.delay(str(extraction.extract_id), job.job_id)
+        job.celery_task_id = result.id
+        job = update_job(session, job)
+
+        return extraction, job

app/services/llm.py

@@ -9,6 +9,14 @@
 logger = get_logger(__name__)
 
 
+def check_ollama_available() -> bool:
+    """Return True if the Ollama server responds with a successful status."""
+    try:
+        return requests.get(f"{OLLAMA_HOST}/api/tags", timeout=3).ok
+    except requests.exceptions.RequestException:
+        return False
+
+
 class LLM:
     def __init__(self, transcript_text: str=None, target_fields: list=None, json_dict: dict=None, model: str=None):
         self._transcript_text = transcript_text

app/tasks/extract.py

@@ -0,0 +1,47 @@
+import logging
+from datetime import datetime, timezone
+from uuid import UUID
+
+from app.api.schemas.enums import ExtractionStatus
+from app.core.celery import celery_app
+from app.db.database import get_session
+from app.db.repositories import get_extraction, get_job_by_uuid, update_extraction, update_job
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(name="extract_incident")
+def extract_task(extract_id_str: str, job_id_str: str) -> dict:
+    """Stub extraction worker.
+
+    #629 ships the queue slice: the extraction row and job exist and the task
+    runs, but the real chunked extraction (split the narrative into field
+    groups, call the model, validate, stitch a contract, create the draft
+    incident) lands in #630. For now the task only marks the job in-flight and
+    leaves the extraction in its ``processing`` state, which is what
+    GET /extract/{id} reports back to pollers.
+    """
+    session = next(get_session())
+    extract_id = UUID(extract_id_str)
+    try:
+        extraction = get_extraction(session, extract_id)
+        job = get_job_by_uuid(session, job_id_str)
+
+        now = datetime.now(timezone.utc)
+        if extraction:
+            extraction.status = ExtractionStatus.processing
+            extraction.started_at = extraction.started_at or now
+            extraction.updated_at = now
+            update_extraction(session, extraction)
+        if job:
+            job.status = "processing"
+            job.updated_at = now
+            update_job(session, job)
+
+        logger.info(
+            "extract_task stub ran for extraction %s; real worker is #630",
+            extract_id_str,
+        )
+        return {"extract_id": extract_id_str, "job_id": job_id_str}
+    finally:
+        session.close()