spike: preserve exam spec-tagging prototype

Persist the paper's question-layout signals (A-level Section A/B, EITHER/OR choice
groups, marks confidence) onto exam_templates.extraction_meta so the setup UI can
show them. Data was carried by the contract but dropped at persistence.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
Claude-Session: https://claude.ai/code/session_01GruxHXxfdp4kZCgAMVFgvV

MET_BY_PRETEST.md

@@ -0,0 +1,7 @@
+Feature: add GET /database/timetable/timetables endpoint for TimetableListPage.
+
+- Added router file: routers/database/timetable/timetables.py
+- Wired new timetable router from run/routers.py
+- Provided pragmatic /database/timetables/timetables GET and GET/{id} shapes that return Timetable lists
+
+Note: TimetableListPage uses /database/timetable/timetables (singular timetable segment) even though the underlying router is now under /database/timetables. Curl checks below use that path.

api/services/docling/extract.py

@@ -16,7 +16,7 @@ question labels from a RapidOCR per-page pass. v2 generalises across exam boards
       per-part marks (N).
   * OCR              <- sequential top-level integers followed by question text, parts (a)/(i),
       marks [N]; `(b)*` flags an extended-response part.
-  * REGIONS          <- Docling layout labels mapped to taxonomy + gemma4:e4b `answer_regions`
+  * REGIONS          <- Docling layout labels mapped to taxonomy + gemma4:e4b-131k `answer_regions`
       (taxonomy #3 — the one structure no deterministic pass emits) merged by part.
   * TABLES           <- Docling `tables` carried through; parts on a table page flagged has_table.
   * COVERAGE         <- recall vs a ground-truth label set: built-in physics GT (regression guard)
@@ -611,7 +611,7 @@ def _norm_region_type(kind):
 
 
 def merge_gemma(parts, gemma_dir):
-    """Attach gemma4:e4b answer_regions (#3) to parts by for_part; gap-fill missing marks."""
+    """Attach gemma4:e4b-131k answer_regions (#3) to parts by for_part; gap-fill missing marks."""
     n_reg = n_fill = 0
     for fn in sorted(glob.glob(os.path.join(gemma_dir, "p*.json"))):
         d = json.load(open(fn))

docker-compose.dev.yml

@@ -55,6 +55,8 @@ services:
       - CC_COMPOSE_SERVICE=backend-dev
       - RUN_INIT=false
       - INIT_MODE=infra
+      # P2: route exam auto-map through the spike's full recognition pipeline (extraction service)
+      - EXAM_EXTRACT_URL=${EXAM_EXTRACT_URL:-http://192.168.0.203:8899}
     ports:
       - "18000:8000"
     depends_on:

modules/services/exam_extract.py

@@ -40,6 +40,19 @@ def _get(base: str, slug: str, timeout: int = 30) -> Dict[str, Any]:
     return r.json()
 
 
+def get_replica(slug: str, timeout: int = 30) -> Dict[str, Any]:
+    """The digital-replica markdown for a paper (P4): {slug, title, n_questions, total_marks, markdown,
+    questions:[{label, marks, markdown}]}. Raises ExtractError if the paper has no replica yet."""
+    base = service_url()
+    if not base:
+        raise ExtractError("EXAM_EXTRACT_URL not configured")
+    r = requests.get(f"{base}/api/replica/{slug}", timeout=timeout)
+    if r.status_code == 404:
+        raise ExtractError(f"no digital replica for {slug}")
+    r.raise_for_status()
+    return r.json()
+
+
 def extract_suggestions(slug: str, pdf_bytes: bytes, *, force: bool = False,
                         poll_timeout: int = 1500, poll_interval: int = 5) -> Dict[str, Any]:
     """POST the paper to the service and poll until the analyse contract is ready.

modules/services/exam_spectag.py

@@ -0,0 +1,123 @@
+"""AI spec-point suggestion (R3.5.3) — classify each question's CROP to its AQA spec topic with a vision LLM.
+
+The app's questions carry geometry (bounds) but no text, so we render the source PDF at the app's 780px
+canvas width (fitz) — the same space the bounds live in — crop each question, and ask a vision model
+(Ollama on the AI host) which topic it assesses, constrained to the paper's subject catalogue. Suggestions
+are written to exam_questions.spec_ref by the caller (only where empty — never overwriting a teacher's tag);
+the teacher confirms + syncs ASSESSES. Validated: qwen3-vl:4b classifies a question crop in ~4s.
+"""
+from __future__ import annotations
+
+import base64
+import io
+import os
+import re
+from typing import Any, Dict, List, Optional, Tuple
+
+import fitz  # PyMuPDF
+import requests
+from PIL import Image
+
+from modules.logger_tool import initialise_logger
+from run.initialization.init_exam_graph import SPECIFICATIONS  # import-safe: no Neo4j connection at import
+
+logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True)
+
+CANVAS_WIDTH = 780  # the app renders the PDF (and emits bounds) at this width
+VISION_MODEL = os.getenv("EXAM_SPECTAG_MODEL", "qwen3-vl:4b")
+
+
+def _ollama_url() -> str:
+    explicit = os.getenv("OLLAMA_URL") or os.getenv("OLLAMA_BASE_URL")
+    if explicit:
+        return explicit.rstrip("/")
+    return f"http://{os.getenv('HOST_OLLAMA', '192.168.0.39')}:{os.getenv('PORT_OLLAMA', '11434')}"
+
+
+def resolve_spec(subject: Optional[str], exam_code: Optional[str]) -> Optional[Dict[str, Any]]:
+    """Find the seeded spec for a paper: match its code digits (e.g. '8463/1' → AQA-PHYS-8463),
+    else a unique subject match. Returns the SPECIFICATIONS entry or None."""
+    digits = re.sub(r"\D", "", exam_code or "")
+    for spec in SPECIFICATIONS:
+        code = re.sub(r"\D", "", spec["spec_code"])
+        if code and code in digits:
+            return spec
+    subj = (subject or "").strip().lower()
+    cands = [s for s in SPECIFICATIONS if s["subject"] == subj]
+    return cands[0] if len(cands) == 1 else None
+
+
+def _render_pages(pdf_bytes: bytes) -> Tuple[List[Image.Image], List[int]]:
+    """Render every page at CANVAS_WIDTH; return (page images, stacked-top y offsets)."""
+    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
+    pages: List[Image.Image] = []
+    tops: List[int] = []
+    acc = 0
+    for page in doc:
+        zoom = CANVAS_WIDTH / page.rect.width
+        pix = page.get_pixmap(matrix=fitz.Matrix(zoom, zoom), alpha=False)
+        pages.append(Image.frombytes("RGB", (pix.width, pix.height), pix.samples))
+        tops.append(acc)
+        acc += pix.height
+    doc.close()
+    return pages, tops
+
+
+def _crop_b64(pages: List[Image.Image], tops: List[int], page: Any, bounds: Dict[str, Any]) -> Optional[str]:
+    try:
+        idx = int(page) - 1
+    except (TypeError, ValueError):
+        return None
+    if idx < 0 or idx >= len(pages):
+        return None
+    img, top = pages[idx], tops[idx]
+    x = max(0, int(bounds.get("x", 0)))
+    y = max(0, int(bounds.get("y", 0)) - top)
+    x2 = min(x + int(bounds.get("w", img.width)), img.width)
+    y2 = min(y + int(bounds.get("h", 80)), img.height)
+    if x2 - x < 3 or y2 - y < 3:
+        return None
+    buf = io.BytesIO()
+    img.crop((x, y, x2, y2)).save(buf, "PNG")
+    return base64.b64encode(buf.getvalue()).decode()
+
+
+def _classify(img_b64: str, spec: Dict[str, Any], valid: set) -> Optional[str]:
+    tlist = "\n".join(f"{ref} {name}" for ref, name in spec["topics"])
+    prompt = (
+        f"This is an AQA {spec['award_code']} {spec['subject'].title()} exam question. Which specification "
+        f"topic does it mainly assess?\n\nTOPICS:\n{tlist}\n\n"
+        f"Answer with ONLY the topic ref from [{' '.join(sorted(valid))}]. Ref only, no words."
+    )
+    body = {"model": VISION_MODEL, "prompt": prompt, "images": [img_b64], "stream": False,
+            "options": {"temperature": 0, "seed": 0}}
+    resp = requests.post(f"{_ollama_url()}/api/generate", json=body, timeout=90)
+    resp.raise_for_status()
+    text = resp.json().get("response", "")
+    for ref in re.findall(r"\d+\.\d+", text):
+        if ref in valid:
+            return ref
+    return None
+
+
+def suggest(subject: Optional[str], exam_code: Optional[str], questions: List[Dict[str, Any]],
+            pdf_bytes: bytes, limit: int = 60) -> Tuple[Dict[str, str], Optional[str]]:
+    """questions = [{id, page, bounds}] already filtered to un-tagged with bounds. Returns ({id: ref}, spec_code)."""
+    spec = resolve_spec(subject, exam_code)
+    if not spec:
+        return {}, None
+    valid = {ref for ref, _ in spec["topics"]}
+    pages, tops = _render_pages(pdf_bytes)
+    out: Dict[str, str] = {}
+    for q in questions[:limit]:
+        b64 = _crop_b64(pages, tops, q.get("page"), q.get("bounds") or {})
+        if not b64:
+            continue
+        try:
+            ref = _classify(b64, spec, valid)
+        except Exception as exc:  # noqa: BLE001 - one bad crop/timeout must not sink the batch
+            logger.warning(f"spec-tag classify failed for question {q.get('id')}: {exc}")
+            continue
+        if ref:
+            out[q["id"]] = ref
+    return out, spec["spec_code"]

routers/exam/schemas.py

@@ -80,6 +80,9 @@ class ResponseAreaPayload(BaseModel):
     ] = None
     # Optional Context differentiation (v1 generic; future graph/chart/data_table/diagram/code_block/passage).
     context_type: Optional[str] = None
+    # Rich recognition payload (75-exam-marker-region-meta.sql): figure name/description, OMR geometry, unit…
+    # Carried on canvas save so a named context figure survives a round-trip.
+    meta: Optional[Dict[str, Any]] = None
     source: Literal["manual", "ai"] = "manual"
     confirmed: bool = True
     confidence: Optional[float] = Field(default=None, ge=0, le=1)

routers/exam/templates.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 import json
 import math
 import os
+import re
 import tempfile
 import time
 import uuid
@@ -30,6 +31,7 @@ from modules.database.services.exam_projection import project_template, project_
 from modules.database.supabase.utils.client import SupabaseServiceRoleClient
 from modules.database.supabase.utils.storage import StorageAdmin
 from modules.upload_validation import read_pdf_upload_bytes
+from modules.services import exam_extract
 from modules.logger_tool import initialise_logger
 from routers.exam.dependencies import ExamContext, get_exam_context, lookup_exam_code
 from routers.exam.schemas import (
@@ -455,6 +457,32 @@ def _y_to_canvas(y_value: float, page_number: int, pages: List[Dict[str, float]]
     return round(g["page_top"] + (g["page_pt_h"] - (float(y_value) - g["crop_y0"])) / g["page_pt_h"] * g["rendered_h"], 2)
 
 
+def _frac_box_to_canvas(bounds: Optional[Dict[str, Any]], page_number: int,
+                        pages: List[Dict[str, float]]) -> Optional[Dict[str, float]]:
+    """Page-fraction {x,y,w,h} (0..1 per page, from the extraction service) → 780-wide stacked canvas."""
+    if not bounds:
+        return None
+    g = _page_geom(pages, page_number)
+    try:
+        x, y, w, h = (float(bounds["x"]), float(bounds["y"]), float(bounds["w"]), float(bounds["h"]))
+    except (KeyError, TypeError, ValueError):
+        return None
+    return {
+        "x": round(x * g["rendered_w"], 2),
+        "y": round(g["page_top"] + y * g["rendered_h"], 2),
+        "w": round(w * g["rendered_w"], 2),
+        "h": round(h * g["rendered_h"], 2),
+    }
+
+
+def _frac_y_to_canvas(y_frac: Any, page_number: int, pages: List[Dict[str, float]]) -> Optional[float]:
+    g = _page_geom(pages, page_number)
+    try:
+        return round(g["page_top"] + float(y_frac) * g["rendered_h"], 2)
+    except (TypeError, ValueError):
+        return None
+
+
 def _ai_id(template_id: str, *parts: Any) -> str:
     return str(uuid.uuid5(uuid.NAMESPACE_URL, "/".join(["cc-auto-map", template_id, *[str(p) for p in parts]])))
 
@@ -660,6 +688,144 @@ def _run_auto_map_job(job_id: str, ctx: ExamContext, template_id: str, pdf_bytes
         _set_auto_map_status(job_id, {"status": "failed", "template_id": template_id, "error": str(exc)})
 
 
+_ALLOWED_RESPONSE_FORMS = {"lines", "answer-box", "working", "diagram", "tick-boxes", "table", "blanks"}
+_ALLOWED_ANSWER_TYPES = {"written", "mcq", "short", "diagram"}
+_ALLOWED_KINDS = {"response", "context", "question_number", "mark_area", "reference", "furniture"}
+_BOARD_RE = re.compile(r"^(aqa|edexcel|ocr|wjec|eduqas|ccea)-")
+
+
+def _extract_slug(ctx: ExamContext, template: Dict[str, Any]) -> str:
+    """A stable, board-prefixed slug for the extraction-service cache. Prefer the catalogue exam_code
+    (e.g. 'AQA-8463-1H-2022JUN-QP' → 'aqa-8463-1h-2022jun-qp' → board 'aqa' for the right margins);
+    fall back to a template-id slug (structure.py then defaults to AQA content-box margins)."""
+    code = None
+    exam_id = template.get("exam_id")
+    if exam_id:
+        try:
+            row = _first(ctx.supabase.table("eb_exams").select("exam_code").eq("id", exam_id).limit(1).execute())
+            code = (row or {}).get("exam_code")
+        except Exception as exc:
+            logger.info(f"extract slug: eb_exams lookup failed for {exam_id}: {exc}")
+    slug = re.sub(r"[^a-z0-9._-]+", "-", (code or "").lower()).strip("-")
+    if _BOARD_RE.match(slug):
+        return slug
+    return f"aqa-tmpl-{str(template.get('id') or '')[:12]}"
+
+
+def _map_service_contract_to_rows(template_id: str, contract: Dict[str, Any],
+                                  pdf_bytes: bytes) -> Dict[str, List[Dict[str, Any]]]:
+    """Map the extraction service's page-fraction analyse contract onto the app's canvas-space ghost rows.
+
+    Coordinates: page-fraction → the 780-wide stacked canvas. IDs: the service's deterministic uuid5s are
+    re-namespaced per template via _ai_id so two templates of the same paper don't collide and a re-map of
+    the same template re-emits stable ids (so _refresh_ai_rows preserves confirmed ghosts). FK-safe: parts
+    whose parent/owner question is absent are de-parented / dropped rather than crashing the insert.
+    """
+    pages = _pdf_page_geometry(pdf_bytes)
+    sug = contract.get("suggestions") or {}
+
+    def qid(uid: Any) -> str:
+        return _ai_id(template_id, "svc-q", uid)
+
+    questions: List[Dict[str, Any]] = []
+    q_ids: set = set()
+    for q in sug.get("questions") or []:
+        uid = q.get("uid")
+        if not uid:
+            continue
+        rid = qid(uid)
+        q_ids.add(rid)
+        at = q.get("answer_type") if q.get("answer_type") in _ALLOWED_ANSWER_TYPES else None
+        questions.append({
+            "id": rid, "template_id": template_id,
+            "parent_id": qid(q["parent_uid"]) if q.get("parent_uid") else None,
+            "label": q.get("label") or "?", "order": q.get("order", len(questions)),
+            "max_marks": _safe_marks(q.get("max_marks")), "answer_type": at,
+            "is_container": bool(q.get("is_container")),
+            "bounds": _frac_box_to_canvas(q.get("bounds"), q.get("page") or 1, pages),
+            "page": q.get("page"), "source": "ai", "confirmed": False,
+            "confidence": _safe_confidence(q.get("confidence")), "derivation": "extract-service",
+            # analyse contract v2 (migration 78): command verb + stem prose per part
+            "command_word": (q.get("command_word") or None) if not q.get("is_container") else None,
+            "preamble": q.get("preamble") or None,
+        })
+    for q in questions:                                 # FK safety: de-parent a dangling parent_id
+        if q["parent_id"] and q["parent_id"] not in q_ids:
+            q["parent_id"] = None
+
+    response_areas: List[Dict[str, Any]] = []
+    for ra in sug.get("response_areas") or []:
+        uid = ra.get("uid")
+        quid = qid(ra.get("question_uid") or "")
+        if not uid or quid not in q_ids:               # orphan region → drop (FK safety)
+            continue
+        bounds = _frac_box_to_canvas(ra.get("bounds"), ra.get("page") or 1, pages)
+        if not bounds:
+            continue
+        kind = ra.get("kind") if ra.get("kind") in _ALLOWED_KINDS else "response"
+        form = ra.get("response_form") if ra.get("response_form") in _ALLOWED_RESPONSE_FORMS else None
+        response_areas.append({
+            "id": _ai_id(template_id, "svc-ra", uid), "template_id": template_id,
+            "question_id": quid, "page": ra.get("page"), "bounds": bounds,
+            "kind": kind, "response_form": form if kind == "response" else None,
+            "context_type": ra.get("context_type"), "meta": ra.get("meta") or {},
+            "source": "ai", "confirmed": False,
+            "confidence": _safe_confidence(ra.get("confidence")), "derivation": "extract-service",
+        })
+
+    boundaries: List[Dict[str, Any]] = []
+    for i, b in enumerate(sug.get("boundaries") or []):
+        quid = qid(b.get("question_uid") or "")
+        if quid not in q_ids:
+            continue
+        page_index = b.get("page_index")
+        y = _frac_y_to_canvas(b.get("y"), (page_index or 0) + 1, pages)
+        if y is None:
+            continue
+        boundaries.append({
+            "id": _ai_id(template_id, "svc-b", b.get("question_uid") or i), "template_id": template_id,
+            "question_id": quid, "label": b.get("label") or "", "page_index": page_index,
+            "y": y, "bounds": None, "source": "ai", "confirmed": False,
+            "confidence": _safe_confidence(b.get("confidence")), "derivation": "extract-service",
+        })
+
+    return {"questions": questions, "response_areas": response_areas, "boundaries": boundaries}
+
+
+def _run_service_extract_merge(ctx: ExamContext, template_id: str, pdf_bytes: bytes, slug: str) -> Dict[str, List[Dict[str, Any]]]:
+    contract = exam_extract.extract_suggestions(slug, pdf_bytes)
+    rows = _map_service_contract_to_rows(template_id, contract, pdf_bytes)
+    _refresh_ai_rows(ctx, template_id, rows)
+    meta = contract.get("meta") or {}
+    # P3: record provenance + the audit cover-reconciliation gate so the setup UI can flag under-reads,
+    # and store the slug so the digital-text view can resolve this paper's replica.
+    updates: Dict[str, Any] = {"extraction_meta": {
+        "engine": "extract-service", "slug": slug, "audit": meta.get("audit") or {},
+        "counts": {k: len(v) for k, v in rows.items()},
+        # question-layout signals for the setup UI: paper sections (e.g. A-level Section A/B) + EITHER/OR choices
+        "sections": meta.get("sections") or [],
+        "choice_groups": meta.get("choice_groups") or [],
+        "marks_confidence": meta.get("marks_confidence"),
+    }}
+    n_pages = meta.get("n_pages") or meta.get("pages")
+    if n_pages:
+        updates["page_count"] = n_pages
+    ctx.supabase.table("exam_templates").update(updates).eq("id", template_id).execute()
+    return rows
+
+
+def _run_service_extract_job(job_id: str, ctx: ExamContext, template_id: str, pdf_bytes: bytes, slug: str) -> None:
+    _set_auto_map_status(job_id, {"status": "running", "template_id": template_id, "engine": "extract-service", "slug": slug})
+    try:
+        rows = _run_service_extract_merge(ctx, template_id, pdf_bytes, slug)
+        project_template_safe(template_id)
+        _set_auto_map_status(job_id, {"status": "completed", "template_id": template_id,
+                                      "engine": "extract-service", "counts": {k: len(v) for k, v in rows.items()}})
+    except Exception as exc:
+        logger.exception(f"extract-service job failed for template {template_id}: {exc}")
+        _set_auto_map_status(job_id, {"status": "failed", "template_id": template_id, "engine": "extract-service", "error": str(exc)})
+
+
 # ─── templates ───────────────────────────────────────────────────────────────
 
 
@@ -837,6 +1003,14 @@ async def auto_map_template(
         raise HTTPException(status_code=409, detail="Template has recorded marks; auto-map structural refresh is blocked.")
     bucket, path, pdf_bytes = _resolve_template_source(ctx, template)
     source_label = f"{bucket}/{path}"
+    # Extraction-service path (P2): when EXAM_EXTRACT_URL is set, route auto-map through the spike's full
+    # recognition pipeline instead of the thin first-pass. Always async — a cold paper is ~15 min.
+    if exam_extract.is_enabled():
+        slug = _extract_slug(ctx, template)
+        job_id = str(uuid.uuid4())
+        _set_auto_map_status(job_id, {"status": "queued", "template_id": template_id, "engine": "extract-service", "slug": slug})
+        background_tasks.add_task(_run_service_extract_job, job_id, ctx, template_id, pdf_bytes, slug)
+        return JSONResponse(status_code=202, content={"status": "accepted", "job_id": job_id, "engine": "extract-service"})
     try:
         fast_path = _pdf_has_text_layer(pdf_bytes)
     except Exception as exc:
@@ -875,6 +1049,24 @@ async def auto_map_status(
     return body
 
 
+@router.get("/templates/{template_id}/digital-text")
+async def template_digital_text(
+    template_id: str,
+    ctx: ExamContext = Depends(get_exam_context),
+) -> Dict[str, Any]:
+    """P4: the digital-replica markdown for this template's paper (stem text, parts, marks, answer-space
+    placeholders, spec refs). Resolves the paper via the slug the extraction used."""
+    template = _fetch_template_or_404(ctx, template_id)
+    _require_source_visibility_or_404(ctx, template)
+    if not exam_extract.is_enabled():
+        raise HTTPException(status_code=503, detail="Extraction service not configured")
+    slug = (template.get("extraction_meta") or {}).get("slug") or _extract_slug(ctx, template)
+    try:
+        return exam_extract.get_replica(slug)
+    except exam_extract.ExtractError as exc:
+        raise HTTPException(status_code=404, detail=f"No digital text yet — run auto-map first ({exc})")
+
+
 @router.put("/templates/{template_id}")
 async def replace_template(
     template_id: str,
@@ -955,6 +1147,7 @@ async def replace_template(
                 "kind": ra.kind,
                 "response_form": ra.response_form,
                 "context_type": ra.context_type,   # 73: optional Context differentiation
+                "meta": ra.meta,                   # 75: rich recognition payload (name/description/OMR/…)
                 "source": ra.source,
                 "confirmed": ra.confirmed,
                 "confidence": ra.confidence,