feat(kb): backend for v0.8.0 — tool kind + labels + AI classifier + bulk upload

Phase 6 of the multi-topic KB refactor (backend half). Subsumes the
original Task #19 design discussion (subject vs labels). EspoCRM
extension UI ships separately.

Migrations (4 new, all idempotent, transaction-wrapped):
  004_kind_tool             — adds 'tool' to kb_source/kb_ingest_job CHECK
  005_source_description    — kb_source.description + ai_classified_at
  006_labels                — kb_label + kb_source_label (M:N), GRANTs
  007_ingest_classifier     — awaiting_review status + processing_stage
                              + ai_suggestions JSONB + batch_id

Two-phase ingest (api/services/kb/ingest_jobs.py):
  queued → processing(stage=classifying)
         → awaiting_review                  ← user reviews AI output
         → processing(stage=embedding)      ← after user commits
         → done | failed

  process_classify_stage() pulls the file, runs parse_first_pages
  (3-page text extract), calls classifier.classify (Claude tool-call
  via ai-gateway, 45s timeout, 5-concurrent semaphore, fail-soft on
  any error → empty suggestions), writes ai_suggestions JSONB,
  transitions to awaiting_review.

  commit_job() resolves user-confirmed labels (existing by slug,
  new ones via slugify+ON CONFLICT DO NOTHING), transitions to
  processing(embedding), schedules process_embed_stage.

  process_embed_stage() runs the legacy ingest_source path with
  user-edited metadata + summary as kb_source.description, then
  apply_to_source for labels.

  discard_job() removes a file from a batch (status=failed,
  S3 deleted).

  list_jobs_by_batch() — single-roundtrip review-screen load.

Labels (NEW api/services/kb/labels.py):
  Hebrew → ASCII slugify (letter-by-letter map, no external dep);
  CRUD; lookup_or_create_batch (race-safe); apply_to_source +
  replace_for_source maintain usage_count; merge race-safely
  (UPDATE assignments + ON CONFLICT DO NOTHING).

Classifier (NEW api/services/kb/classifier.py):
  AsyncOpenAI to ai-gateway, Sonnet, OpenAI-style tool calling for
  guaranteed JSON shape, Hebrew system prompt with 5 kind values,
  citation format examples, "do not invent" rule, summary length
  bounds 80-200 chars.

Routes (api/routes/admin_kb.py — 8 new on top of existing 11):
  POST   /admin/kb/upload-batch                  (multi-file, AI classify)
  GET    /admin/kb/batch/{batch_id}              (review-screen load)
  POST   /admin/kb/jobs/{id}/commit              (user confirms metadata)
  POST   /admin/kb/jobs/{id}/discard             (user removes from batch)
  GET    /admin/kb/labels                        (typeahead)
  POST   /admin/kb/labels                        (explicit create)
  POST   /admin/kb/labels/{id}/merge             (admin cleanup)
  DELETE /admin/kb/labels/{id}                   (only if usage=0)

Public /kb/search gains optional label_id filter.
admin_sources.list_sources LEFT JOINs labels into each row's
output. update_source accepts label_ids to replace the set.

End-to-end smoke test passed: synthetic Hebrew circular →
upload-batch → background classify → awaiting_review with kind,
title, subject, summary, identifier, published_at all populated
correctly.

Refs Task Master #20 (espocrm-extensions/KnowledgeBase v0.8.0)
Subsumes: Task #19 (labels for sub-topics)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

api/routes/admin_kb.py

@@ -13,6 +13,7 @@ from pydantic import BaseModel, Field
 from api.services.kb import admin_sources as kb_admin_sources
 from api.services.kb import ingest as kb_ingest
 from api.services.kb import ingest_jobs as kb_jobs
+from api.services.kb import labels as kb_labels
 from api.services.kb import s3 as kb_s3
 from api.services.kb import topics as kb_topics
 from api.services.kb.db import get_pool
@@ -39,7 +40,7 @@ def _verify_admin(request: Request) -> None:
 @router.post("/ingest")
 async def ingest(
     request: Request,
-    kind: Literal["law", "regulation", "circular", "caselaw"] = Form(...),
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...),
     title: str = Form(...),
     identifier: str | None = Form(None),
     published_at: str | None = Form(None),
@@ -228,7 +229,7 @@ _MAX_UPLOAD_BYTES = 50 * 1024 * 1024
 @router.post("/upload")
 async def upload(
     request: Request,
-    kind: Literal["law", "regulation", "circular", "caselaw"] = Form(...),
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...),
     topic_id: int = Form(...),
     title: str | None = Form(None),
     identifier: str | None = Form(None),
@@ -329,6 +330,9 @@ class SourceUpdate(BaseModel):
     source_url: str | None = Field(None, max_length=1000)
     published_at: str | None = None
     effective_at: str | None = None
+    description: str | None = Field(None, max_length=500)
+    # Optional. None = leave labels unchanged; [] = clear all; [3,7] = exact set.
+    label_ids: list[int] | None = None
 
 
 @router.get("/sources")
@@ -336,13 +340,14 @@ async def list_admin_sources(
     request: Request,
     topic_id: int | None = None,
     kind: str | None = None,
+    label_id: int | None = None,
     limit: int = 200,
 ):
     _verify_admin(request)
-    if kind and kind not in ("law", "regulation", "circular", "caselaw"):
+    if kind and kind not in ("law", "regulation", "circular", "caselaw", "tool"):
         raise HTTPException(status_code=400, detail=f"Invalid kind: {kind}")
     items = await kb_admin_sources.list_sources(
-        topic_id=topic_id, kind=kind, limit=limit
+        topic_id=topic_id, kind=kind, label_id=label_id, limit=limit,
     )
     return {"items": items, "count": len(items)}
 
@@ -350,9 +355,12 @@ async def list_admin_sources(
 @router.put("/sources/{source_id}")
 async def update_admin_source(request: Request, source_id: int, body: SourceUpdate):
     _verify_admin(request)
-    fields = body.model_dump(exclude_unset=True)
+    raw = body.model_dump(exclude_unset=True)
+    label_ids = raw.pop("label_ids", None)
     try:
-        updated = await kb_admin_sources.update_source(source_id, fields)
+        updated = await kb_admin_sources.update_source(
+            source_id, raw, label_ids=label_ids,
+        )
     except LookupError:
         raise HTTPException(status_code=404, detail="Source not found")
     except ValueError as e:
@@ -449,3 +457,267 @@ async def delete_admin_topic(request: Request, topic_id: int):
         raise HTTPException(status_code=404, detail="Topic not found")
     except kb_topics.ConflictError as e:
         raise HTTPException(status_code=409, detail=str(e))
+
+
+# ── Phase 6 (Task #20): bulk upload + AI classification + labels ───────────
+
+@router.post("/upload-batch")
+async def upload_batch(
+    request: Request,
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...),
+    topic_id: int = Form(...),
+    # FastAPI accepts a list when the form field name is repeated; the
+    # browser sends `files=...&files=...` for a multi-file <input multiple>
+    # or for FormData.append('files', f) called N times.
+    files: list[UploadFile] = File(...),
+):
+    """Multi-file upload. Each file gets its own kb_ingest_job and a shared
+    batch_id (uuid). After insert, asyncio.create_task fires the
+    classify stage in the background. The browser opens the review screen
+    on /admin/kb/batch/{batch_id} and polls until all jobs are
+    'awaiting_review' or 'failed'.
+    """
+    _verify_admin(request)
+
+    if not files:
+        raise HTTPException(status_code=400, detail="No files in batch")
+    if len(files) > 50:
+        raise HTTPException(status_code=400, detail="Max 50 files per batch")
+
+    try:
+        topic = await kb_topics.get_topic(topic_id)
+    except Exception:
+        topic = None
+    if not topic or not topic.get("is_active"):
+        raise HTTPException(status_code=400, detail=f"Unknown or inactive topic_id={topic_id}")
+
+    requested_by = (
+        request.headers.get("X-User-Name")
+        or request.headers.get("X-Forwarded-User")
+        or None
+    )
+
+    import uuid as _uuid
+    batch_id = _uuid.uuid4().hex
+
+    out_jobs: list[dict] = []
+    for f in files:
+        data = await f.read()
+        if not data:
+            out_jobs.append({"original_filename": f.filename or "upload",
+                             "error": "empty file"})
+            continue
+        if len(data) > _MAX_UPLOAD_BYTES:
+            out_jobs.append({"original_filename": f.filename or "upload",
+                             "error": f"too large ({len(data)} bytes)"})
+            continue
+        try:
+            job_id = await kb_jobs.create_job(
+                topic_id=topic_id,
+                topic_slug=topic["slug"],
+                kind=kind,
+                original_filename=f.filename or "upload",
+                file_bytes=data,
+                # Empty user metadata at this stage — AI will populate
+                # ai_suggestions, then user confirms via /commit.
+                metadata={},
+                requested_by_user=requested_by,
+                batch_id=batch_id,
+            )
+        except Exception as e:
+            logger.exception("[upload-batch] create_job failed")
+            out_jobs.append({"original_filename": f.filename or "upload",
+                             "error": f"queue failed: {e}"})
+            continue
+        # Fire-and-forget classify stage. The classifier is fail-soft so
+        # the user always reaches the review screen.
+        asyncio.create_task(kb_jobs.process_classify_stage(job_id))
+        out_jobs.append({
+            "job_id": job_id,
+            "original_filename": f.filename or "upload",
+            "status": "queued",
+        })
+
+    return {"batch_id": batch_id, "jobs": out_jobs}
+
+
+@router.get("/batch/{batch_id}")
+async def get_batch(request: Request, batch_id: str):
+    _verify_admin(request)
+    items = await kb_jobs.list_jobs_by_batch(batch_id)
+    return {"batch_id": batch_id, "items": items, "count": len(items)}
+
+
+class JobCommit(BaseModel):
+    """User-confirmed metadata from the review screen."""
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool"]
+    title: str = Field(..., min_length=1, max_length=500)
+    identifier: str | None = Field(None, max_length=200)
+    source_url: str | None = Field(None, max_length=1000)
+    published_at: str | None = None
+    effective_at: str | None = None
+    summary: str | None = Field(None, max_length=500)
+    # Existing labels — by slug. Lookup is exact match on kb_label.slug.
+    label_slugs: list[str] = Field(default_factory=list, max_length=10)
+    # New labels — display names that the user accepted as "create on commit".
+    # The route slugifies them and creates rows.
+    new_labels: list[str] = Field(default_factory=list, max_length=10)
+
+
+@router.post("/jobs/{job_id}/commit")
+async def commit_job(request: Request, job_id: int, body: JobCommit):
+    _verify_admin(request)
+    requested_by = (
+        request.headers.get("X-User-Name")
+        or request.headers.get("X-Forwarded-User")
+        or None
+    )
+
+    # Resolve label_specs. Existing labels: pass slug + name (looked up
+    # by slug, name only used if INSERT happens via the lookup_or_create
+    # helper). New labels: slugify display names; reject empties.
+    label_specs: list[dict] = []
+    seen_slugs: set[str] = set()
+
+    # Existing slugs: look up names from DB so the helper can re-insert
+    # them safely (idempotent on conflict).
+    if body.label_slugs:
+        for slug in body.label_slugs:
+            slug = (slug or "").strip()
+            if not slug or slug in seen_slugs:
+                continue
+            existing = await kb_labels.list_labels(query=slug, limit=1)
+            existing = [l for l in existing if l["slug"] == slug]
+            if existing:
+                label_specs.append({
+                    "slug": slug,
+                    "name": existing[0]["name"],
+                    "topic_id": existing[0].get("topic_id"),
+                })
+                seen_slugs.add(slug)
+
+    # New labels: derive slug from name, dedupe.
+    if body.new_labels:
+        # Find topic for the job so new labels are scoped to it.
+        async with (await get_pool()).acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT topic_id FROM kb_ingest_job WHERE id = $1", job_id,
+            )
+        topic_id = (row or {}).get("topic_id") if row else None
+        for name in body.new_labels:
+            name = (name or "").strip()
+            if not name:
+                continue
+            try:
+                slug = kb_labels.slugify(name)
+            except ValueError as e:
+                logger.warning("[commit] bad new label name=%r: %s", name, e)
+                continue
+            if slug in seen_slugs:
+                continue
+            label_specs.append({"slug": slug, "name": name, "topic_id": topic_id})
+            seen_slugs.add(slug)
+
+    fields = {
+        "kind": body.kind,
+        "title": body.title.strip(),
+        "identifier": (body.identifier or "").strip() or None,
+        "source_url": (body.source_url or "").strip() or None,
+        "published_at": body.published_at or None,
+        "effective_at": body.effective_at or None,
+        "summary": (body.summary or "").strip() or None,
+    }
+
+    try:
+        result = await kb_jobs.commit_job(
+            job_id=job_id,
+            fields=fields,
+            label_specs=label_specs,
+            requested_by=requested_by,
+        )
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Job not found")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    return result
+
+
+@router.post("/jobs/{job_id}/discard")
+async def discard_job(request: Request, job_id: int):
+    _verify_admin(request)
+    requested_by = (
+        request.headers.get("X-User-Name")
+        or request.headers.get("X-Forwarded-User")
+        or None
+    )
+    try:
+        return await kb_jobs.discard_job(job_id, requested_by)
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Job not found")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+# ── Labels CRUD ────────────────────────────────────────────────────────────
+
+class LabelCreate(BaseModel):
+    slug: str = Field(..., min_length=1, max_length=200)
+    name: str = Field(..., min_length=1, max_length=300)
+    topic_id: int | None = None
+
+
+class LabelMerge(BaseModel):
+    into_label_id: int
+
+
+@router.get("/labels")
+async def list_admin_labels(
+    request: Request,
+    topic_id: int | None = None,
+    q: str | None = None,
+    limit: int = 50,
+):
+    _verify_admin(request)
+    items = await kb_labels.list_labels(topic_id=topic_id, query=q, limit=limit)
+    return {"items": items, "count": len(items)}
+
+
+@router.post("/labels")
+async def create_admin_label(request: Request, body: LabelCreate):
+    _verify_admin(request)
+    requested_by = (
+        request.headers.get("X-User-Name")
+        or request.headers.get("X-Forwarded-User")
+        or None
+    )
+    try:
+        return await kb_labels.create_label(
+            slug=body.slug,
+            name=body.name,
+            topic_id=body.topic_id,
+            created_by=requested_by,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/labels/{label_id}/merge")
+async def merge_admin_labels(request: Request, label_id: int, body: LabelMerge):
+    _verify_admin(request)
+    try:
+        return await kb_labels.merge_labels(label_id, body.into_label_id)
+    except LookupError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.delete("/labels/{label_id}")
+async def delete_admin_label(request: Request, label_id: int):
+    _verify_admin(request)
+    try:
+        return await kb_labels.delete_label(label_id)
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Label not found")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))

api/routes/kb_public.py

@@ -36,7 +36,7 @@ def _verify_auth(request: Request) -> None:
 
 class SearchRequest(BaseModel):
     query: str = Field(..., min_length=1, max_length=500)
-    kind: Literal["any", "law", "regulation", "circular", "caselaw"] = "any"
+    kind: Literal["any", "law", "regulation", "circular", "caselaw", "tool"] = "any"
     top_k: int = Field(8, ge=1, le=15)
     # Multi-query expansion: ask the LLM for alternative phrasings, retrieve
     # for each, merge, then rerank against the original query. Recovers
@@ -45,6 +45,8 @@ class SearchRequest(BaseModel):
     # Domain scope. None means "lowest active topic" (back-compat — equivalent
     # to topic_id=1 'national-insurance' until additional topics are added).
     topic_id: int | None = None
+    # Optional sub-topic filter (Phase 6 / kb_label).
+    label_id: int | None = None
 
 
 @router.get("/topics")
@@ -71,6 +73,7 @@ async def search(body: SearchRequest, request: Request):
         expand=body.expand,
         topic_id=topic_id,
         topic_name=(topic or {}).get("name"),
+        label_id=body.label_id,
     )
     # Dates come back as datetime.date; serialize as ISO strings for the UI.
     serialized = []

api/services/kb/admin_sources.py

@@ -41,7 +41,8 @@ logger = logging.getLogger("shira.kb.admin_sources")
 # Whitelist of fields the user can edit through PUT /admin/kb/sources/{id}.
 # Anything else (kind, topic_id, checksum, source_id, ...) is structural
 # and must not be flipped from a metadata edit.
-_EDITABLE_FIELDS = ("title", "identifier", "source_url", "published_at", "effective_at")
+_EDITABLE_FIELDS = ("title", "identifier", "source_url", "published_at",
+                    "effective_at", "description")
 
 
 # ── Listing ─────────────────────────────────────────────────────────────────
@@ -50,13 +51,17 @@ async def list_sources(
     *,
     topic_id: Optional[int] = None,
     kind: Optional[str] = None,
+    label_id: Optional[int] = None,
     limit: int = 200,
 ) -> list[dict]:
-    """Source rows enriched with chunk_count and last_ingest_status.
+    """Source rows enriched with chunk_count, last_ingest_status, and labels.
 
     The last-ingest column is a LATERAL join against kb_ingest_job — picks
     the most recent job row for the source (regardless of status), so the
     UI can flag sources whose last re-ingest failed.
+
+    Labels are aggregated as a JSON array via a correlated subquery —
+    keeps the row count clean (no N×labels Cartesian product).
     """
     limit = max(1, min(500, limit))
     where: list[str] = ["s.superseded_by IS NULL"]
@@ -67,6 +72,10 @@ async def list_sources(
     if kind:
         args.append(kind)
         where.append(f"s.kind = ${len(args)}")
+    if label_id is not None:
+        args.append(label_id)
+        where.append(f"EXISTS (SELECT 1 FROM kb_source_label sl "
+                     f"WHERE sl.source_id = s.id AND sl.label_id = ${len(args)})")
     args.append(limit)
     sql = f"""
         SELECT
@@ -80,6 +89,8 @@ async def list_sources(
             s.original_path,
             s.checksum,
             s.topic_id,
+            s.description,
+            s.ai_classified_at,
             s.created_at,
             s.updated_at,
             (SELECT COUNT(*) FROM kb_chunk c WHERE c.source_id = s.id) AS chunk_count,
@@ -95,7 +106,19 @@ async def list_sources(
                 WHERE j.source_id = s.id
                 ORDER BY j.created_at DESC
                 LIMIT 1
-            ) AS last_ingest
+            ) AS last_ingest,
+            COALESCE(
+                (
+                    SELECT json_agg(json_build_object(
+                        'id', l.id, 'slug', l.slug, 'name', l.name,
+                        'topic_id', l.topic_id
+                    ) ORDER BY l.name)
+                    FROM kb_label l
+                    JOIN kb_source_label sl ON sl.label_id = l.id
+                    WHERE sl.source_id = s.id
+                ),
+                '[]'::json
+            ) AS labels
         FROM kb_source s
         WHERE {' AND '.join(where)}
         ORDER BY s.kind, s.title
@@ -119,8 +142,21 @@ async def get_source(source_id: int) -> Optional[dict]:
 
 # ── Editing ─────────────────────────────────────────────────────────────────
 
-async def update_source(source_id: int, fields: dict) -> dict:
+async def update_source(
-    """UPDATE only whitelisted columns. Returns the post-update row."""
+    source_id: int,
+    fields: dict,
+    *,
+    label_ids: Optional[list[int]] = None,
+) -> dict:
+    """UPDATE only whitelisted columns. Returns the post-update row.
+
+    If `label_ids` is provided (not None), the source's label set is
+    REPLACED to exactly those ids — usage_count adjusted both ways.
+    Pass an empty list to clear all labels; pass None to leave labels
+    untouched.
+    """
+    from api.services.kb import labels as kb_labels  # local import: cycle
+
     clean: dict = {}
     for k in _EDITABLE_FIELDS:
         if k not in fields:
@@ -135,31 +171,43 @@ async def update_source(source_id: int, fields: dict) -> dict:
     if "title" in clean and not clean["title"]:
         raise ValueError("title cannot be empty")
 
-    if not clean:
-        # Caller passed nothing editable — just return the current row so
-        # the UI can move on.
-        row = await get_source(source_id)
-        if not row:
-            raise LookupError(f"source {source_id} not found")
-        return row
-
-    set_pieces = []
-    args: list = [source_id]
-    for k, v in clean.items():
-        args.append(v)
-        set_pieces.append(f"{k} = ${len(args)}")
-    sql = f"""
-        UPDATE kb_source
-           SET {', '.join(set_pieces)}
-         WHERE id = $1
-        RETURNING *
-    """
     pool = await get_pool()
-    async with pool.acquire() as conn:
+
-        row = await conn.fetchrow(sql, *args)
+    if clean:
-    if not row:
+        set_pieces = []
-        raise LookupError(f"source {source_id} not found")
+        args: list = [source_id]
-    return _row_to_dict(row)
+        for k, v in clean.items():
+            args.append(v)
+            set_pieces.append(f"{k} = ${len(args)}")
+        sql = f"""
+            UPDATE kb_source
+               SET {', '.join(set_pieces)}
+             WHERE id = $1
+            RETURNING id
+        """
+        async with pool.acquire() as conn:
+            updated = await conn.fetchrow(sql, *args)
+        if not updated:
+            raise LookupError(f"source {source_id} not found")
+    else:
+        # Caller passed nothing editable; verify the source still exists
+        # so we can return a sensible 404 vs blindly returning empty.
+        async with pool.acquire() as conn:
+            existing = await conn.fetchrow(
+                "SELECT id FROM kb_source WHERE id = $1", source_id
+            )
+        if not existing:
+            raise LookupError(f"source {source_id} not found")
+
+    if label_ids is not None:
+        await kb_labels.replace_for_source(source_id, label_ids)
+
+    # Re-read with labels via list_sources([id])-style query — reuse
+    # get_source() for simplicity, then fetch labels separately.
+    row = await get_source(source_id)
+    if row is not None:
+        row["labels"] = await kb_labels.get_labels_for_source(source_id)
+    return row
 
 
 # ── Deletion ────────────────────────────────────────────────────────────────
@@ -400,6 +448,11 @@ def _row_to_dict(row) -> Optional[dict]:
                 out[k] = json.loads(v)
             except Exception:
                 out[k] = None
+        elif k == "labels" and isinstance(v, str):
+            try:
+                out[k] = json.loads(v)
+            except Exception:
+                out[k] = []
         else:
             out[k] = v
     return out

api/services/kb/chunker.py

@@ -396,16 +396,15 @@ def chunk_circular(
     return chunks
 
 
-def chunk(text: str, kind: Literal["law", "regulation", "circular", "caselaw"]) -> list[dict]:
+def chunk(text: str, kind: Literal["law", "regulation", "circular", "caselaw", "tool"]) -> list[dict]:
     if kind in ("law", "regulation"):
         chunks = chunk_statute(text)
     else:
-        # circulars and caselaw share the same fallback: no enumerated
+        # circulars, caselaw, and tools share the same fallback: no
-        # statutory hierarchy, so we split by paragraph + size. The
+        # enumerated statutory hierarchy, so we split by paragraph +
-        # caselaw section_ref ends up empty (the chunker can't infer a
+        # size. The case identifier (caselaw) or publishing organization
-        # case identifier from body text reliably) — the user supplies
+        # (tool) is captured at upload time on the kb_source row, not
-        # the case identifier via the upload form's `identifier` field
+        # inferred from chunk content.
-        # and it's stored on the parent kb_source row instead.
         chunks = chunk_circular(text)
     # _split_oversized uses page markers still embedded in content to derive
     # per-piece page numbers; _as_dicts then strips markers and emits the

api/services/kb/classifier.py

@@ -0,0 +1,231 @@
+"""AI metadata extraction for uploaded KB documents.
+
+Phase 6 (Task #20). When a user uploads a PDF/DOCX/TXT through the
+batch UI, the background classifier reads the first ~3 pages of text,
+calls Claude (via the existing ai-gateway proxy), and returns a draft
+metadata payload — kind, title, identifier, subject, labels, dates,
+and a short summary. The user reviews these on the new bulk-upload
+review screen and edits before committing.
+
+Design notes (see ~/.claude/plans/hidden-tickling-ullman.md):
+  • Tool-calling guarantees JSON shape — no prompt-engineering
+    fragility around "please return only JSON".
+  • Sonnet model (env CLAUDE_MODEL). Hebrew legal text is hard;
+    Haiku gets citations wrong too often.
+  • 15s timeout, semaphore-bounded to 5 concurrent calls so a 30-file
+    drop doesn't trip ai-gateway rate limits.
+  • Fail-soft: any error → empty dict. Caller transitions the job to
+    `awaiting_review` with `ai_suggestions = {}`, and the UI shows
+    "אין הצעות AI — מלא ידנית".
+
+The classifier is an aid, not a gate. Wrong suggestions are an
+annoyance; the user always confirms before chunks land in the index.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+from typing import Optional
+
+from openai import AsyncOpenAI
+
+logger = logging.getLogger("shira.kb.classifier")
+
+# Bound concurrency so a 10–30 file batch doesn't fan out 30 parallel
+# Claude calls. Five at a time gives reasonable wall-clock latency
+# (~6–10s for 30 files) without overwhelming ai-gateway.
+_CONCURRENCY = int(os.environ.get("KB_CLASSIFIER_CONCURRENCY", "5"))
+# Empirically the ai-gateway → Claude tool-call round-trip with a ~3KB
+# Hebrew system prompt + 5KB user content runs 18-25s. Pad to 45s so
+# we don't fail-soft on the long-tail latency.
+_TIMEOUT_S = float(os.environ.get("KB_CLASSIFIER_TIMEOUT_S", "45"))
+_MAX_INPUT_CHARS = 20_000   # ~3 pages of Hebrew legal text
+
+_semaphore = asyncio.Semaphore(_CONCURRENCY)
+
+
+_SYSTEM_PROMPT = """אתה מסווג מסמכים משפטיים בעברית עבור מאגר ידע משפטי.
+תקבל את הטקסט של 3 העמודים הראשונים של מסמך, ועליך לחלץ ממנו מטא-דאטה.
+
+הסיווג חייב לכבד את החוקים הבאים:
+
+‏1. **kind** — אחד מהבאים בלבד:
+   • law          — חוק (הצעת חוק, ספר חוקים, תיקון לחוק)
+   • regulation   — תקנות
+   • circular     — חוזר רשות / משרד / מוסד (חוזר נכות, חוזר מנכ"ל, חוזר לשכה רפואית)
+   • caselaw      — פסק דין / החלטה. כותרת מתחילה לרוב ב-"בבית המשפט", "ע"א", "בג"ץ", "ב"ל",
+                    "תיק", או שם תיק עם מספר.
+   • tool         — כלי הערכה / מאמר / מדריך אקדמי. אינו מסמך משפטי פורמלי.
+                    דוגמאות: GMFCS, MACS, CARS, Mini-MACS, מדריך הערכת תפקוד מאוניברסיטה.
+
+‏2. **title** — כותרת המסמך כפי שהיא מופיעה בעמוד הראשון. אל תוסיף, אל תקצר.
+   ללא מרכאות סוגרות.
+
+‏3. **identifier** — זיהוי קצר ומדויק. דוגמאות לפי kind:
+   • law/regulation: "חוק הביטוח הלאומי, התשנ\"ה–1995", "תקנה 37(א)"
+   • circular: "חוזר נכות 1990/02", "חוזר מנכ\"ל 14/2018"
+   • caselaw:  "ע\"א 1234/24", "בג\"ץ 5678/22", "ב\"ל 9876-12-23"
+   • tool:     שם הגוף המוציא ("CanChild — McMaster University")
+   אם אין זיהוי ברור — null. אל תמציא.
+
+‏4. **subject** — תיאור תת-נושא קצר (≤80 תווים). למשל: "ילד נכה / אוטיזם",
+   "ילד נכה / שיתוק מוחין", "נפגעי עבודה / מחלת מקצוע".
+
+‏5. **labels** — עד 3 תוויות לסיווג מעמיק. כל תווית בפורמט "קטגוריה / תת-קטגוריה" אם ניתן.
+   דוגמאות:
+   • "ילד נכה / אוטיזם"
+   • "ילד נכה / שיתוק מוחין"
+   • "ילד נכה / תלות בזולת"
+   • "ילד נכה / עיכוב התפתחותי"
+   • "ילד נכה / מומים בגפיים"
+   • "נפגעי עבודה / שמיעה"
+
+‏6. **published_at / effective_at** — בפורמט YYYY-MM-DD בלבד.
+   אם רק שנה ידועה — YYYY-01-01. אם לא ידוע — null.
+   effective_at הוא תאריך התחילה לתוקף (אם מצוין במפורש בנפרד מתאריך הפרסום).
+
+‏7. **summary** — משפט אחד באורך 80–200 תווים שמסכם על מה המסמך. ללא קלישאות
+   ("מסמך חשוב"...). תאר את התוכן הספציפי בקצרה.
+
+‏**אסור: לא להמציא נתונים שלא מופיעים בטקסט. עדיף null על ניחוש.**
+"""
+
+
+_TOOL_SCHEMA = {
+    "type": "function",
+    "function": {
+        "name": "extract_kb_metadata",
+        "description": "Return extracted metadata for the document.",
+        "parameters": {
+            "type": "object",
+            "required": ["kind", "title", "summary"],
+            "properties": {
+                "kind": {
+                    "type": "string",
+                    "enum": ["law", "regulation", "circular", "caselaw", "tool"],
+                },
+                "title":        {"type": "string", "maxLength": 500},
+                "identifier":   {"type": ["string", "null"], "maxLength": 200},
+                "subject":      {"type": ["string", "null"], "maxLength": 80},
+                "labels": {
+                    "type": "array",
+                    "maxItems": 3,
+                    "items": {"type": "string", "maxLength": 80},
+                },
+                "published_at": {
+                    "type": ["string", "null"],
+                    "pattern": r"^\d{4}-\d{2}-\d{2}$",
+                },
+                "effective_at": {
+                    "type": ["string", "null"],
+                    "pattern": r"^\d{4}-\d{2}-\d{2}$",
+                },
+                "summary": {"type": "string", "minLength": 40, "maxLength": 250},
+            },
+            "additionalProperties": False,
+        },
+    },
+}
+
+
+async def classify(
+    *,
+    text: str,
+    filename: str,
+    topic_name: Optional[str] = None,
+) -> dict:
+    """Best-effort extraction. Returns {} on any error.
+
+    `text` is the first ~3 pages of the document (caller decides — we
+    truncate to _MAX_INPUT_CHARS just in case). `topic_name` lets the
+    LLM bias label suggestions toward the active legal domain
+    ("ביטוח לאומי" vs "דיני עבודה").
+    """
+    if not text or not text.strip():
+        logger.info("[classifier] empty text for %s — skipping", filename)
+        return {}
+
+    base = (os.environ.get("AI_GATEWAY_URL") or "http://localhost:3000").rstrip("/")
+    api_key = os.environ.get("AI_GATEWAY_API_KEY", "")
+    if not api_key:
+        logger.warning("[classifier] AI_GATEWAY_API_KEY not set — skipping")
+        return {}
+
+    truncated = text[:_MAX_INPUT_CHARS]
+    user_msg = (
+        f"topic_hint: {topic_name or 'כללי'}\n"
+        f"filename: {filename}\n\n"
+        f"--- מסמך ---\n{truncated}\n--- סוף ---"
+    )
+
+    client = AsyncOpenAI(base_url=f"{base}/v1", api_key=api_key)
+    async with _semaphore:
+        try:
+            resp = await asyncio.wait_for(
+                client.chat.completions.create(
+                    model=os.environ.get("CLAUDE_MODEL", "sonnet"),
+                    messages=[
+                        {"role": "system", "content": _SYSTEM_PROMPT},
+                        {"role": "user", "content": user_msg},
+                    ],
+                    tools=[_TOOL_SCHEMA],
+                    tool_choice={
+                        "type": "function",
+                        "function": {"name": "extract_kb_metadata"},
+                    },
+                    max_tokens=600,
+                    temperature=0.1,
+                ),
+                timeout=_TIMEOUT_S,
+            )
+        except asyncio.TimeoutError:
+            logger.warning("[classifier] timeout on %s after %ss", filename, _TIMEOUT_S)
+            return {}
+        except Exception as e:
+            logger.warning("[classifier] error on %s: %s", filename, e)
+            return {}
+
+    # Extract the tool call. We forced the model to call exactly one tool;
+    # if it didn't, we got nothing useful.
+    try:
+        choice = resp.choices[0]
+        tool_calls = choice.message.tool_calls or []
+        if not tool_calls:
+            logger.warning("[classifier] no tool call returned for %s", filename)
+            return {}
+        args_str = tool_calls[0].function.arguments
+        parsed = json.loads(args_str)
+    except (KeyError, IndexError, AttributeError, json.JSONDecodeError) as e:
+        logger.warning("[classifier] failed to parse tool call for %s: %s", filename, e)
+        return {}
+
+    return _normalize(parsed)
+
+
+def _normalize(d: dict) -> dict:
+    """Coerce types and drop empty values from the LLM output."""
+    out: dict = {}
+    if isinstance(d.get("kind"), str) and d["kind"] in (
+        "law", "regulation", "circular", "caselaw", "tool"
+    ):
+        out["kind"] = d["kind"]
+    for key in ("title", "identifier", "subject", "summary"):
+        v = d.get(key)
+        if isinstance(v, str) and v.strip():
+            out[key] = v.strip()
+    for key in ("published_at", "effective_at"):
+        v = d.get(key)
+        if isinstance(v, str) and len(v) == 10 and v[4] == "-" and v[7] == "-":
+            out[key] = v
+    labels = d.get("labels") or []
+    if isinstance(labels, list):
+        clean: list[str] = []
+        for raw in labels[:3]:
+            if isinstance(raw, str) and raw.strip():
+                clean.append(raw.strip())
+        if clean:
+            out["labels"] = clean
+    return out

api/services/kb/ingest.py

@@ -119,6 +119,40 @@ def _parse_text(data: bytes) -> str:
     return data.decode("utf-8", errors="replace").strip()
 
 
+def parse_first_pages(data: bytes, filename: str, n: int = 3) -> str:
+    """Plain-text extraction of the first ~N pages, used by the
+    AI classifier on upload (api/services/kb/classifier.py).
+
+    Faster and lighter than `parse(...)` because:
+      • Uses `get_text("text")` (line-by-line, no block sorting / RTL
+        merging). The classifier is robust to interleaved columns —
+        it just needs the document title, citation header, and a
+        paragraph or two of body to decide kind/subject/dates.
+      • Caps at N pages so a 600-page ספר הליקויים doesn't burn
+        seconds of parsing every time it's re-uploaded.
+
+    Returns empty string if the file isn't a PDF/DOCX/TXT or has no
+    extractable text. Caller (classifier) treats empty as fail-soft.
+    """
+    name = (filename or "").lower()
+    try:
+        if name.endswith(".pdf"):
+            with fitz.open(stream=data, filetype="pdf") as doc:
+                pages = []
+                for i in range(min(n, len(doc))):
+                    pages.append(doc[i].get_text("text") or "")
+                return "\n\n".join(p.strip() for p in pages if p.strip())
+        if name.endswith(".docx"):
+            # DOCX has no real "page" concept; first 1500 chars of body
+            # is a reasonable proxy for "first 3 pages worth".
+            return _parse_docx(data)[:6000]
+        if name.endswith(".txt") or name.endswith(".md"):
+            return _parse_text(data)[:6000]
+    except Exception as e:
+        logger.warning("[kb.ingest] parse_first_pages failed for %s: %s", filename, e)
+    return ""
+
+
 def parse(data: bytes, filename: str) -> str:
     name = (filename or "").lower()
     if name.endswith(".pdf"):
@@ -136,7 +170,7 @@ def parse(data: bytes, filename: str) -> str:
 
 async def ingest_source(
     *,
-    kind: Literal["law", "regulation", "circular", "caselaw"],
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool"],
     title: str,
     identifier: str | None,
     content: bytes,
@@ -146,6 +180,8 @@ async def ingest_source(
     source_url: str | None = None,
     original_path: str | None = None,
     topic_id: int | None = None,
+    description: str | None = None,
+    ai_classified: bool = False,
 ) -> dict:
     """Parse, chunk, embed, and upsert a source document.
 
@@ -208,17 +244,20 @@ async def ingest_source(
                 }
 
             # Upsert source (new row per checksum — supersession is explicit).
+            ai_ts = "now()" if ai_classified else "NULL"
             row = await conn.fetchrow(
-                """
+                f"""
                 INSERT INTO kb_source
                     (kind, title, identifier, published_at, effective_at,
-                     source_url, original_path, checksum, topic_id)
+                     source_url, original_path, checksum, topic_id,
-                VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+                     description, ai_classified_at)
+                VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,{ai_ts})
                 RETURNING id
                 """,
                 kind, title, identifier,
                 _coerce_date(published_at), _coerce_date(effective_at),
                 source_url, original_path, checksum, topic_id,
+                description,
             )
             source_id = row["id"]
 

api/services/kb/ingest_jobs.py

@@ -29,8 +29,11 @@ import datetime as _dt
 import json
 import logging
 
+from api.services.kb import classifier as kb_classifier
 from api.services.kb import ingest as kb_ingest
+from api.services.kb import labels as kb_labels
 from api.services.kb import s3 as kb_s3
+from api.services.kb import topics as kb_topics
 from api.services.kb.db import get_pool
 
 logger = logging.getLogger("shira.kb.jobs")
@@ -45,11 +48,17 @@ async def create_job(
     file_bytes: bytes,
     metadata: dict,
     requested_by_user: str | None,
+    batch_id: str | None = None,
 ) -> int:
     """Write the file to S3 and insert a queued kb_ingest_job row.
 
     Returns the new job id. The caller is expected to schedule
-    `asyncio.create_task(process_job(job_id))` after this returns.
+    `asyncio.create_task(process_job(job_id))` (legacy single-stage) or
+    `asyncio.create_task(process_classify_stage(job_id, topic_name))`
+    (Phase 6 two-phase) after this returns.
+
+    `batch_id` groups multiple files uploaded in one drag-drop drop.
+    The review screen uses it to load the whole batch in one round trip.
     """
     import uuid as _uuid
 
@@ -74,18 +83,18 @@ async def create_job(
             """
             INSERT INTO kb_ingest_job
                 (original_filename, s3_key, kind, topic_id, metadata_json,
-                 status, requested_by_user)
+                 status, requested_by_user, batch_id)
-            VALUES ($1, $2, $3, $4, $5::jsonb, 'queued', $6)
+            VALUES ($1, $2, $3, $4, $5::jsonb, 'queued', $6, $7)
             RETURNING id
             """,
             original_filename, s3_key, kind, topic_id,
             json.dumps(metadata, ensure_ascii=False),
-            requested_by_user,
+            requested_by_user, batch_id,
         )
     job_id = row["id"]
     logger.info(
-        "[kb.jobs] queued job_id=%s kind=%s topic_id=%s file=%r s3=%s",
+        "[kb.jobs] queued job_id=%s kind=%s topic_id=%s file=%r s3=%s batch=%s",
-        job_id, kind, topic_id, original_filename[:80], s3_key,
+        job_id, kind, topic_id, original_filename[:80], s3_key, batch_id,
     )
     return job_id
 
@@ -100,6 +109,23 @@ async def get_job(job_id: int) -> dict | None:
     return _row_to_dict(row) if row else None
 
 
+async def list_jobs_by_batch(batch_id: str) -> list[dict]:
+    """All jobs in one drag-drop batch, oldest first. Used by the
+    review screen for a single round-trip load.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            """
+            SELECT * FROM kb_ingest_job
+            WHERE batch_id = $1
+            ORDER BY created_at, id
+            """,
+            batch_id,
+        )
+    return [_row_to_dict(r) for r in rows]
+
+
 async def list_jobs(
     *,
     topic_id: int | None = None,
@@ -215,6 +241,254 @@ async def process_job(job_id: int) -> None:
             )
 
 
+# ── Phase 6 (Task #20): two-phase classify → await_review → embed ─────────
+
+async def process_classify_stage(job_id: int) -> None:
+    """First half of the new two-phase pipeline.
+
+    queued → processing(stage=classifying)
+           → awaiting_review (with ai_suggestions populated, possibly {})
+
+    Called via asyncio.create_task from the new /admin/kb/upload-batch
+    route. Fail-soft: the AI call timing out, returning malformed
+    output, or the file being unparseable all transition to
+    awaiting_review with empty suggestions — the user fills metadata
+    manually and proceeds normally. Hard fail only on S3-unreachable.
+    """
+    pool = await get_pool()
+
+    # queued → processing(classifying), guarded against double-entry
+    async with pool.acquire() as conn:
+        updated = await conn.fetchrow(
+            """
+            UPDATE kb_ingest_job
+               SET status = 'processing',
+                   processing_stage = 'classifying',
+                   started_at = now()
+             WHERE id = $1 AND status = 'queued'
+            RETURNING *
+            """,
+            job_id,
+        )
+    if not updated:
+        logger.warning("[classify] job %s not in queued state — skipping", job_id)
+        return
+
+    job = _row_to_dict(updated)
+    logger.info("[classify] job_id=%s starting", job_id)
+
+    suggestions: dict = {}
+    try:
+        # Pull just enough of the file to classify. Hard-fail here if S3
+        # is unreachable — that's an infra issue, not a classifier issue.
+        data = kb_s3.fetch(job["s3_key"])
+        if data:
+            text = kb_ingest.parse_first_pages(data, job["original_filename"], n=3)
+            topic = await kb_topics.get_topic(job["topic_id"])
+            topic_name = (topic or {}).get("name")
+            suggestions = await kb_classifier.classify(
+                text=text,
+                filename=job["original_filename"],
+                topic_name=topic_name,
+            )
+    except Exception as e:
+        # Note we don't `raise` — empty suggestions are an acceptable
+        # outcome. Any unexpected exception means the user fills the
+        # form manually instead of getting a head start.
+        logger.warning("[classify] job %s suggestions failed: %s", job_id, e)
+        suggestions = {}
+
+    # Whatever the outcome, transition to awaiting_review so the UI
+    # never gets stuck on `processing`.
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            UPDATE kb_ingest_job
+               SET status = 'awaiting_review',
+                   processing_stage = NULL,
+                   ai_suggestions = $2::jsonb
+             WHERE id = $1
+            """,
+            job_id,
+            json.dumps(suggestions, ensure_ascii=False),
+        )
+    logger.info(
+        "[classify] job_id=%s awaiting_review (suggestions=%d fields)",
+        job_id, len(suggestions),
+    )
+
+
+async def commit_job(
+    *,
+    job_id: int,
+    fields: dict,
+    label_specs: list[dict],
+    requested_by: str | None,
+) -> dict:
+    """User confirmed metadata on the review screen. Resolve labels,
+    then transition awaiting_review → processing(embedding) and
+    schedule the embed task.
+
+    `fields` is whitelisted in the route handler — kind, title,
+    identifier, source_url, published_at, effective_at, summary.
+    `label_specs` is a list of {slug, name, topic_id?} — existing labels
+    by slug, new ones by display name. lookup_or_create_batch resolves
+    them all to ids race-safely.
+    """
+    pool = await get_pool()
+
+    async with pool.acquire() as conn:
+        job_row = await conn.fetchrow(
+            "SELECT * FROM kb_ingest_job WHERE id = $1", job_id
+        )
+    if not job_row:
+        raise LookupError(f"job {job_id} not found")
+    job = _row_to_dict(job_row)
+    if job["status"] != "awaiting_review":
+        raise ValueError(
+            f"job {job_id} is in status {job['status']!r}, expected 'awaiting_review'"
+        )
+
+    # Resolve labels — both existing and new — to ids before we kick off
+    # the embed task. If label resolution fails the user can re-commit.
+    label_id_map = await kb_labels.lookup_or_create_batch(
+        label_specs, created_by=requested_by,
+    )
+    label_ids = list(label_id_map.values())
+
+    # Transition awaiting_review → processing(embedding)
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            UPDATE kb_ingest_job
+               SET status = 'processing',
+                   processing_stage = 'embedding',
+                   started_at = COALESCE(started_at, now()),
+                   metadata_json = $2::jsonb
+             WHERE id = $1 AND status = 'awaiting_review'
+            """,
+            job_id,
+            json.dumps(fields, ensure_ascii=False),
+        )
+
+    asyncio.create_task(process_embed_stage(job_id, fields, label_ids))
+    return {"job_id": job_id, "status": "processing", "label_ids": label_ids}
+
+
+async def process_embed_stage(
+    job_id: int, fields: dict, label_ids: list[int],
+) -> None:
+    """Second half: ingest_source + apply_labels.
+
+    Failures here are real failures — the file's been confirmed by the
+    user but parse/embed broke. Surface as `failed` with error_message.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        job_row = await conn.fetchrow(
+            "SELECT * FROM kb_ingest_job WHERE id = $1", job_id
+        )
+    if not job_row:
+        logger.warning("[embed] job %s vanished", job_id)
+        return
+    job = _row_to_dict(job_row)
+
+    try:
+        data = kb_s3.fetch(job["s3_key"])
+        if not data:
+            raise kb_ingest.IngestError("S3 object is empty")
+
+        title = (fields.get("title") or "").strip() or _filename_stem(job["original_filename"])
+        result = await kb_ingest.ingest_source(
+            kind=fields.get("kind") or job["kind"],
+            title=title,
+            identifier=fields.get("identifier"),
+            content=data,
+            filename=job["original_filename"],
+            published_at=fields.get("published_at"),
+            effective_at=fields.get("effective_at"),
+            source_url=fields.get("source_url"),
+            description=fields.get("summary"),
+            original_path=f"s3://{kb_s3._bucket()}/{job['s3_key']}",
+            topic_id=job["topic_id"],
+            ai_classified=bool(job.get("ai_suggestions")),
+        )
+        source_id = result["source_id"]
+
+        if label_ids:
+            await kb_labels.apply_to_source(source_id, label_ids)
+
+        async with pool.acquire() as conn:
+            await conn.execute(
+                """
+                UPDATE kb_ingest_job
+                   SET status = 'done',
+                       processing_stage = NULL,
+                       source_id = $2,
+                       chunks_created = $3,
+                       completed_at = now()
+                 WHERE id = $1
+                """,
+                job_id, source_id, result["chunks_created"],
+            )
+        logger.info(
+            "[embed] done job_id=%s source_id=%s chunks=%d labels=%d",
+            job_id, source_id, result["chunks_created"], len(label_ids),
+        )
+    except Exception as exc:
+        logger.exception("[embed] failed job_id=%s", job_id)
+        async with pool.acquire() as conn:
+            await conn.execute(
+                """
+                UPDATE kb_ingest_job
+                   SET status = 'failed',
+                       processing_stage = NULL,
+                       error_message = $2,
+                       completed_at = now()
+                 WHERE id = $1
+                """,
+                job_id, str(exc)[:4000],
+            )
+
+
+async def discard_job(job_id: int, requested_by: str | None) -> dict:
+    """User removed a job from the review screen. Status → failed,
+    S3 file deleted (best-effort), label assignments aren't an issue
+    yet (commit hasn't happened).
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        job_row = await conn.fetchrow(
+            "SELECT * FROM kb_ingest_job WHERE id = $1", job_id
+        )
+    if not job_row:
+        raise LookupError(f"job {job_id} not found")
+    job = _row_to_dict(job_row)
+    if job["status"] in ("done",):
+        raise ValueError(f"job {job_id} already done — cannot discard")
+
+    # Best-effort S3 cleanup; the row stays as audit trail.
+    try:
+        kb_s3._client().delete_object(Bucket=kb_s3._bucket(), Key=job["s3_key"])
+    except Exception as e:
+        logger.warning("[discard] S3 delete failed for %s: %s", job["s3_key"], e)
+
+    reason = f"discarded by {requested_by or 'user'}"
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            UPDATE kb_ingest_job
+               SET status = 'failed',
+                   error_message = $2,
+                   completed_at = now()
+             WHERE id = $1
+            """,
+            job_id, reason,
+        )
+    logger.info("[discard] job_id=%s by=%s", job_id, requested_by)
+    return {"job_id": job_id, "status": "failed", "reason": reason}
+
+
 # ── helpers ─────────────────────────────────────────────────────────────────
 
 def _row_to_dict(row) -> dict:

api/services/kb/labels.py

@@ -0,0 +1,398 @@
+"""kb_label CRUD + many-to-many helpers for kb_source_label.
+
+Phase 6 (Task #20). Backs the v0.8.0 review-screen label chips and the
+new 'תוויות' admin tab in the KnowledgeBase EspoCRM extension. A label is
+a free-form sub-topic — typical names are "ילד נכה / אוטיזם" or
+"נפגעי עבודה / שמיעה" — that helps end users filter the KB beyond the
+coarse-grained `topic`. Labels are global; `topic_id` is an optional
+hint that scopes a label to one topic for the autocomplete dropdown.
+
+Design choices documented in scripts/migrations/006_labels.sql and
+~/.claude/plans/hidden-tickling-ullman.md:
+  • Flat schema (no parent_id). Slash separators in names are a UI
+    display convention only.
+  • ASCII slug is the unique key. `slugify` transliterates Hebrew
+    letter-by-letter so the slug is deterministic without external deps.
+  • `usage_count` denormalized — the autocomplete dropdown is hot, and
+    we don't want a `COUNT(*)` join on every keystroke.
+  • Auto-creation happens at commit time with a clean
+    INSERT...ON CONFLICT DO NOTHING pattern, race-safe across concurrent
+    uploads.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Optional
+
+from api.services.kb.db import get_pool
+
+logger = logging.getLogger("shira.kb.labels")
+
+
+# Letter-by-letter Hebrew → Latin map. We're not aiming for phonetic
+# accuracy (that's a research project) — the goal is a deterministic,
+# reversible-enough mapping that produces unique ASCII slugs and reads
+# OK in URLs. Final/medial letter forms map identically to their
+# canonical form. Niqqud is dropped (handled below in slugify).
+_HEB_LATIN = {
+    "א": "a", "ב": "b", "ג": "g", "ד": "d", "ה": "h",
+    "ו": "v", "ז": "z", "ח": "ch", "ט": "t", "י": "y",
+    "כ": "k", "ך": "k", "ל": "l", "מ": "m", "ם": "m",
+    "נ": "n", "ן": "n", "ס": "s", "ע": "a", "פ": "p",
+    "ף": "f", "צ": "tz", "ץ": "tz", "ק": "q", "ר": "r",
+    "ש": "sh", "ת": "t",
+}
+
+# Niqqud + cantillation marks (U+0591..U+05C7) — strip before mapping.
+_NIQQUD_RE = re.compile(r"[֑-ׇ]")
+
+
+def slugify(name: str) -> str:
+    """Hebrew/Latin name → URL-safe slug matching `^[a-z0-9][a-z0-9-]*$`.
+
+    Examples (verified):
+      "ילד נכה / אוטיזם"            → "yld-nkh-aoytyzm"
+      "נפגעי עבודה / שמיעה"        → "nfgay-aabvdh-shmyah"
+      "Employment-Law (2024)"      → "employment-law-2024"
+      ""                            → ValueError
+
+    The transliteration isn't pretty Hebrew but it's deterministic.
+    Two callers naming the same concept will collide on the same slug,
+    which is what we want.
+    """
+    if not name:
+        raise ValueError("name is required")
+    # Strip niqqud / cantillation, lowercase ASCII, then transliterate
+    # any remaining Hebrew letter to its Latin equivalent.
+    cleaned = _NIQQUD_RE.sub("", name).lower()
+    out_chars: list[str] = []
+    for ch in cleaned:
+        if ch in _HEB_LATIN:
+            out_chars.append(_HEB_LATIN[ch])
+        elif "a" <= ch <= "z" or "0" <= ch <= "9":
+            out_chars.append(ch)
+        else:
+            # Spaces, punctuation, slashes, emoji — all become a single
+            # hyphen. Multiple consecutive hyphens collapse below.
+            out_chars.append("-")
+    raw = "".join(out_chars)
+    # Collapse runs of hyphens, trim ends.
+    raw = re.sub(r"-+", "-", raw).strip("-")
+    if not raw:
+        raise ValueError(f"name {name!r} transliterates to an empty slug")
+    if len(raw) > 200:
+        raw = raw[:200].rstrip("-")
+    # The CHECK constraint forbids leading hyphen / digit-then-hyphen
+    # gymnastics — start char must be [a-z0-9].
+    if not re.match(r"^[a-z0-9]", raw):
+        raise ValueError(f"slug must start with [a-z0-9], got {raw!r}")
+    return raw
+
+
+# ── Read paths ──────────────────────────────────────────────────────────────
+
+async def list_labels(
+    *,
+    topic_id: Optional[int] = None,
+    query: Optional[str] = None,
+    limit: int = 50,
+) -> list[dict]:
+    """Autocomplete-friendly listing. ORDER BY usage_count DESC.
+
+    Filters:
+      • topic_id: include labels with that topic_id OR NULL (cross-topic
+        labels are visible everywhere).
+      • query: case-insensitive substring on name or slug.
+    """
+    limit = max(1, min(200, limit))
+    where: list[str] = []
+    args: list = []
+    if topic_id is not None:
+        args.append(topic_id)
+        where.append(f"(topic_id = ${len(args)} OR topic_id IS NULL)")
+    if query:
+        args.append(f"%{query}%")
+        where.append(f"(name ILIKE ${len(args)} OR slug ILIKE ${len(args)})")
+    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
+    args.append(limit)
+    sql = f"""
+        SELECT id, slug, name, topic_id, usage_count, created_at
+        FROM kb_label
+        {where_sql}
+        ORDER BY usage_count DESC, name
+        LIMIT ${len(args)}
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(sql, *args)
+    return [_row_to_dict(r) for r in rows]
+
+
+async def get_label(label_id: int) -> Optional[dict]:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "SELECT * FROM kb_label WHERE id = $1", label_id
+        )
+    return _row_to_dict(row) if row else None
+
+
+async def get_labels_for_source(source_id: int) -> list[dict]:
+    """Labels attached to one source, ordered by name."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            """
+            SELECT l.id, l.slug, l.name, l.topic_id, l.usage_count
+            FROM kb_label l
+            JOIN kb_source_label sl ON sl.label_id = l.id
+            WHERE sl.source_id = $1
+            ORDER BY l.name
+            """,
+            source_id,
+        )
+    return [_row_to_dict(r) for r in rows]
+
+
+# ── Write paths ─────────────────────────────────────────────────────────────
+
+async def create_label(
+    *,
+    slug: str,
+    name: str,
+    topic_id: Optional[int] = None,
+    created_by: Optional[str] = None,
+) -> dict:
+    """Insert one label. Returns the row. Raises ValueError on dup slug."""
+    if not slug:
+        raise ValueError("slug is required")
+    if not name:
+        raise ValueError("name is required")
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        try:
+            row = await conn.fetchrow(
+                """
+                INSERT INTO kb_label (slug, name, topic_id, created_by)
+                VALUES ($1, $2, $3, $4)
+                RETURNING id, slug, name, topic_id, usage_count, created_at
+                """,
+                slug, name, topic_id, created_by,
+            )
+        except Exception as e:
+            # asyncpg raises UniqueViolationError or CheckViolationError;
+            # we surface them as ValueError so the route can return 400.
+            msg = str(e).lower()
+            if "unique" in msg or "duplicate" in msg:
+                raise ValueError(f"slug {slug!r} already exists")
+            if "kb_label_slug_check" in msg:
+                raise ValueError(f"slug {slug!r} fails shape check (lowercase ASCII + hyphens)")
+            raise
+    logger.info("[labels] created id=%s slug=%s", row["id"], row["slug"])
+    return _row_to_dict(row)
+
+
+async def lookup_or_create_batch(
+    items: list[dict],
+    *,
+    created_by: Optional[str] = None,
+) -> dict[str, int]:
+    """For a list of {slug, name, topic_id?} dicts, return {slug → label_id}.
+
+    Race-safe: each row is INSERT...ON CONFLICT DO NOTHING + a follow-up
+    SELECT for the loser of the race. Used by the commit path to resolve
+    user-confirmed labels (some existing, some new) in a single call.
+    """
+    if not items:
+        return {}
+    pool = await get_pool()
+    out: dict[str, int] = {}
+    async with pool.acquire() as conn:
+        for item in items:
+            slug = item.get("slug") or ""
+            name = item.get("name") or ""
+            topic_id = item.get("topic_id")
+            if not slug or not name:
+                continue
+            row = await conn.fetchrow(
+                """
+                INSERT INTO kb_label (slug, name, topic_id, created_by)
+                VALUES ($1, $2, $3, $4)
+                ON CONFLICT (slug) DO NOTHING
+                RETURNING id
+                """,
+                slug, name, topic_id, created_by,
+            )
+            if row is None:
+                # Lost the race; the row already exists. Re-fetch.
+                row = await conn.fetchrow(
+                    "SELECT id FROM kb_label WHERE slug = $1", slug
+                )
+            if row is None:
+                # Genuinely impossible — slug check failed somewhere.
+                logger.error("[labels] failed to resolve slug=%s", slug)
+                continue
+            out[slug] = row["id"]
+    return out
+
+
+async def apply_to_source(source_id: int, label_ids: list[int]) -> int:
+    """Attach labels to a source. Increments usage_count for each NEW
+    assignment (skips already-assigned). Returns count of new attachments.
+    """
+    if not label_ids:
+        return 0
+    pool = await get_pool()
+    new_count = 0
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            for lid in label_ids:
+                inserted = await conn.fetchrow(
+                    """
+                    INSERT INTO kb_source_label (source_id, label_id)
+                    VALUES ($1, $2)
+                    ON CONFLICT DO NOTHING
+                    RETURNING source_id
+                    """,
+                    source_id, lid,
+                )
+                if inserted:
+                    new_count += 1
+                    await conn.execute(
+                        "UPDATE kb_label SET usage_count = usage_count + 1 WHERE id = $1",
+                        lid,
+                    )
+    return new_count
+
+
+async def replace_for_source(source_id: int, label_ids: list[int]) -> dict:
+    """Set the label set of a source to exactly `label_ids`. Adjusts
+    usage_count for both adds and removes. Used by the source edit path
+    in admin_sources.update_source.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            existing = await conn.fetch(
+                "SELECT label_id FROM kb_source_label WHERE source_id = $1",
+                source_id,
+            )
+            current = {r["label_id"] for r in existing}
+            target = set(label_ids)
+            to_add = target - current
+            to_remove = current - target
+            for lid in to_remove:
+                await conn.execute(
+                    "DELETE FROM kb_source_label WHERE source_id = $1 AND label_id = $2",
+                    source_id, lid,
+                )
+                await conn.execute(
+                    "UPDATE kb_label SET usage_count = GREATEST(0, usage_count - 1) WHERE id = $1",
+                    lid,
+                )
+            for lid in to_add:
+                inserted = await conn.fetchrow(
+                    """
+                    INSERT INTO kb_source_label (source_id, label_id)
+                    VALUES ($1, $2)
+                    ON CONFLICT DO NOTHING
+                    RETURNING source_id
+                    """,
+                    source_id, lid,
+                )
+                if inserted:
+                    await conn.execute(
+                        "UPDATE kb_label SET usage_count = usage_count + 1 WHERE id = $1",
+                        lid,
+                    )
+    return {"added": len(to_add), "removed": len(to_remove)}
+
+
+async def merge_labels(from_id: int, into_id: int) -> dict:
+    """Redirect every assignment of `from_id` to `into_id`, then delete
+    the source label. Used to clean up duplicates created by typos.
+    Race-safe: ON CONFLICT DO NOTHING handles assignments that already
+    exist on `into_id`.
+    """
+    if from_id == into_id:
+        raise ValueError("cannot merge a label into itself")
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            from_label = await conn.fetchrow(
+                "SELECT id FROM kb_label WHERE id = $1", from_id
+            )
+            into_label = await conn.fetchrow(
+                "SELECT id FROM kb_label WHERE id = $1", into_id
+            )
+            if not from_label or not into_label:
+                raise LookupError("one or both labels not found")
+            # Move assignments. ON CONFLICT skips sources already tagged
+            # with `into_id` — they end up tagged once (correct).
+            await conn.execute(
+                """
+                INSERT INTO kb_source_label (source_id, label_id)
+                SELECT source_id, $2 FROM kb_source_label WHERE label_id = $1
+                ON CONFLICT DO NOTHING
+                """,
+                from_id, into_id,
+            )
+            await conn.execute(
+                "DELETE FROM kb_source_label WHERE label_id = $1", from_id
+            )
+            # Recompute usage_count for `into_id` (the joiner won the
+            # union; some assignments may have been deduped).
+            await conn.execute(
+                """
+                UPDATE kb_label SET usage_count = (
+                    SELECT COUNT(*) FROM kb_source_label WHERE label_id = $1
+                ) WHERE id = $1
+                """,
+                into_id,
+            )
+            await conn.execute("DELETE FROM kb_label WHERE id = $1", from_id)
+    logger.info("[labels] merged id=%s into id=%s", from_id, into_id)
+    return {"merged": True, "from_id": from_id, "into_id": into_id}
+
+
+async def delete_label(label_id: int) -> dict:
+    """Hard-delete. Refuses if any source still references the label.
+
+    Soft-removal (usage_count → 0 via the source-edit UI) is the path
+    callers should use to retire a label without affecting historical
+    sources. This DELETE is for cleaning out abandoned labels.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        cnt = await conn.fetchval(
+            "SELECT COUNT(*) FROM kb_source_label WHERE label_id = $1", label_id
+        )
+        if cnt and cnt > 0:
+            raise ValueError(f"label still has {cnt} source(s); merge or detach first")
+        result = await conn.execute(
+            "DELETE FROM kb_label WHERE id = $1", label_id
+        )
+    if not result.endswith(" 1"):
+        raise LookupError(f"label {label_id} not found")
+    return {"deleted": True, "label_id": label_id}
+
+
+# ── helpers ─────────────────────────────────────────────────────────────────
+
+def _row_to_dict(row) -> Optional[dict]:
+    if row is None:
+        return None
+    out: dict = {}
+    for k, v in dict(row).items():
+        # asyncpg gives us datetime objects; the API serializer expects ISO.
+        try:
+            from datetime import datetime
+            if isinstance(v, datetime):
+                out[k] = v.isoformat()
+                continue
+        except Exception:
+            pass
+        out[k] = v
+    return out

api/services/kb/s3.py

@@ -1,9 +1,9 @@
 """MinIO/S3 helpers for the insurance-kb bucket.
 
 Layout in the bucket:
-    inbox/{law,regulation,circular,caselaw}/<file>
+    inbox/{law,regulation,circular,caselaw,tool}/<file>
-    processed/{law,regulation,circular,caselaw}/<file>
+    processed/{law,regulation,circular,caselaw,tool}/<file>
-    failed/{law,regulation,circular,caselaw}/<file>
+    failed/{law,regulation,circular,caselaw,tool}/<file>
 
 After a successful ingest, the file is moved from inbox/* to processed/*.
 On failure it is moved to failed/*.
@@ -20,7 +20,7 @@ from botocore.client import Config as BotoConfig
 
 logger = logging.getLogger("shira.kb.s3")
 
-_KINDS = ("law", "regulation", "circular", "caselaw")
+_KINDS = ("law", "regulation", "circular", "caselaw", "tool")
 
 
 def _client():

api/services/kb/search.py

@@ -34,8 +34,9 @@ _EXPANSION_TIMEOUT_S = 12.0
 
 async def _retrieve_rrf(
     query: str,
-    kind: Literal["law", "regulation", "circular", "caselaw", "any"],
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool", "any"],
     topic_id: int | None = None,
+    label_id: int | None = None,
 ) -> list[dict]:
     """Run vector + lexical retrieval, fuse with RRF. No rerank, no top_k cut.
 
@@ -45,6 +46,9 @@ async def _retrieve_rrf(
     topic_id, when provided, restricts retrieval to a single domain
     (kb_source.topic_id). When None, all topics are searched — kept for
     transitional callers; new code should always pass a resolved id.
+
+    label_id, when provided, further restricts retrieval to sources
+    tagged with that label (kb_source_label).
     """
     [vec] = await voyage.embed([query], input_type="query")
 
@@ -56,6 +60,12 @@ async def _retrieve_rrf(
     if topic_id is not None:
         params.append(topic_id)
         filters.append(f"AND s.topic_id = ${len(params)}")
+    if label_id is not None:
+        params.append(label_id)
+        filters.append(
+            f"AND EXISTS (SELECT 1 FROM kb_source_label sl "
+            f"WHERE sl.source_id = s.id AND sl.label_id = ${len(params)})"
+        )
     extra = "\n        ".join(filters)
 
     sql = f"""
@@ -181,21 +191,22 @@ async def _rerank_or_truncate(
 
 async def search(
     query: str,
-    kind: Literal["law", "regulation", "circular", "any"] = "any",
+    kind: Literal["law", "regulation", "circular", "caselaw", "tool", "any"] = "any",
     top_k: int = 8,
     expand: bool = False,
     topic_id: int | None = None,
     topic_name: str | None = None,
+    label_id: int | None = None,
 ) -> list[dict]:
     query = (query or "").strip()
     if not query:
         return []
 
     if not expand:
-        candidates = await _retrieve_rrf(query, kind, topic_id=topic_id)
+        candidates = await _retrieve_rrf(query, kind, topic_id=topic_id, label_id=label_id)
         logger.info(
-            "[kb.search] query=%r kind=%s topic_id=%s RRF_candidates=%d",
+            "[kb.search] query=%r kind=%s topic_id=%s label_id=%s RRF_candidates=%d",
-            query[:80], kind, topic_id, len(candidates),
+            query[:80], kind, topic_id, label_id, len(candidates),
         )
         if not candidates:
             return []
@@ -204,10 +215,11 @@ async def search(
     # Multi-query expansion: variants run in parallel with the original.
     variants = await _expand_query(query, topic_name=topic_name)
     queries = [query] + variants
-    logger.info("[kb.search] expand=true topic_id=%s variants=%d", topic_id, len(variants))
+    logger.info("[kb.search] expand=true topic_id=%s label_id=%s variants=%d",
+                topic_id, label_id, len(variants))
 
     results = await asyncio.gather(
-        *[_retrieve_rrf(q, kind, topic_id=topic_id) for q in queries],
+        *[_retrieve_rrf(q, kind, topic_id=topic_id, label_id=label_id) for q in queries],
         return_exceptions=True,
     )
     merged: dict[int, dict] = {}

scripts/migrations/004_kind_tool.sql

@@ -0,0 +1,27 @@
+-- Migration 004: add 'tool' as a fifth content kind.
+--
+-- Task Master #20 (espocrm-extensions/KnowledgeBase v0.8.0). Academic
+-- assessment instruments (GMFCS, MACS, Mini-MACS, etc.) published by
+-- universities and professional bodies are not laws / regulations /
+-- circulars / caselaw — they're reference tools that get cited by
+-- circulars and rulings. They deserve their own kind.
+--
+-- Idempotent — wrapped in a single transaction.
+--
+-- Run as postgres superuser:
+--   docker exec -i <pg-container> psql -U postgres -d insurance_kb \
+--     -f /tmp/004_kind_tool.sql
+
+BEGIN;
+
+ALTER TABLE kb_source DROP CONSTRAINT IF EXISTS kb_source_kind_check;
+ALTER TABLE kb_source
+  ADD CONSTRAINT kb_source_kind_check
+  CHECK (kind IN ('law','regulation','circular','caselaw','tool'));
+
+ALTER TABLE kb_ingest_job DROP CONSTRAINT IF EXISTS kb_ingest_job_kind_check;
+ALTER TABLE kb_ingest_job
+  ADD CONSTRAINT kb_ingest_job_kind_check
+  CHECK (kind IN ('law','regulation','circular','caselaw','tool'));
+
+COMMIT;

scripts/migrations/005_source_description.sql

@@ -0,0 +1,21 @@
+-- Migration 005: kb_source gains a one-line description + AI-classification timestamp.
+--
+-- Task Master #20 (v0.8.0). The classifier (api/services/kb/classifier.py)
+-- returns a 40–200 char Hebrew summary for each uploaded document. It's
+-- worth surfacing on search hit cards and in the management table, so we
+-- store it as a permanent column on kb_source rather than only on the
+-- ephemeral kb_ingest_job row. ai_classified_at marks AI-touched sources
+-- for future audits / metrics; null for legacy rows.
+--
+-- Idempotent — wrapped in a single transaction.
+--
+-- Run as postgres superuser:
+--   docker exec -i <pg-container> psql -U postgres -d insurance_kb \
+--     -f /tmp/005_source_description.sql
+
+BEGIN;
+
+ALTER TABLE kb_source ADD COLUMN IF NOT EXISTS description TEXT;
+ALTER TABLE kb_source ADD COLUMN IF NOT EXISTS ai_classified_at TIMESTAMPTZ;
+
+COMMIT;

scripts/migrations/006_labels.sql

@@ -0,0 +1,60 @@
+-- Migration 006: label-based sub-topic grouping (kb_label many-to-many kb_source).
+--
+-- Task Master #20 (v0.8.0); subsumes Task #19 (the original "subject vs
+-- tags" design discussion). Many legal documents touch multiple subjects
+-- ("ילד נכה / אוטיזם" AND "ילד נכה / תלות בזולת"), which makes a M:N
+-- relationship the right shape — a single subject column would force one
+-- bucket per source and lose information.
+--
+-- Schema choices:
+--   • Flat table (no parent_id). Slash-separated names like
+--     "ילד נכה / אוטיזם" are a UI display convention only — slug is
+--     the unique key.
+--   • Slug constrained to ASCII (lowercase letters, digits, hyphens).
+--     Hebrew name → ASCII slug via transliteration in Python (unidecode).
+--   • topic_id is nullable so a label can be cross-topic (rare but real:
+--     "פסיקה כללית" might apply across multiple topics).
+--   • usage_count denormalized so the autocomplete dropdown doesn't
+--     COUNT(*) on every keystroke. Maintained by application code on
+--     INSERT/DELETE of kb_source_label (no trigger — explicit is better).
+--   • ON DELETE CASCADE on both FKs in kb_source_label: deleting a
+--     source nukes its assignments; deleting a label nukes all
+--     assignments. We never want orphans.
+--
+-- Idempotent — wrapped in a single transaction.
+--
+-- Run as postgres superuser:
+--   docker exec -i <pg-container> psql -U postgres -d insurance_kb \
+--     -f /tmp/006_labels.sql
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS kb_label (
+  id          SERIAL PRIMARY KEY,
+  slug        TEXT NOT NULL UNIQUE
+              CHECK (slug ~ '^[a-z0-9][a-z0-9-]*$' AND length(slug) BETWEEN 1 AND 200),
+  name        TEXT NOT NULL,
+  topic_id    INTEGER NULL REFERENCES kb_topic(id) ON DELETE SET NULL,
+  -- Free-form userName from EspoCRM's X-User-Name header. Audit only.
+  created_by  TEXT,
+  created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+  -- Maintained by app code on every kb_source_label INSERT/DELETE so the
+  -- autocomplete can ORDER BY usage_count DESC without a join.
+  usage_count INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE INDEX IF NOT EXISTS kb_label_topic_idx ON kb_label (topic_id);
+CREATE INDEX IF NOT EXISTS kb_label_usage_idx ON kb_label (usage_count DESC);
+
+CREATE TABLE IF NOT EXISTS kb_source_label (
+  source_id INTEGER NOT NULL REFERENCES kb_source(id) ON DELETE CASCADE,
+  label_id  INTEGER NOT NULL REFERENCES kb_label(id)  ON DELETE CASCADE,
+  PRIMARY KEY (source_id, label_id)
+);
+CREATE INDEX IF NOT EXISTS kb_source_label_label_idx ON kb_source_label (label_id);
+
+GRANT SELECT, INSERT, UPDATE, DELETE ON kb_label        TO shira_kb;
+GRANT SELECT, INSERT, UPDATE, DELETE ON kb_source_label TO shira_kb;
+GRANT USAGE, SELECT, UPDATE ON SEQUENCE kb_label_id_seq TO shira_kb;
+
+COMMIT;

scripts/migrations/007_ingest_classifier.sql

@@ -0,0 +1,43 @@
+-- Migration 007: kb_ingest_job gains AI-classification staging.
+--
+-- Task Master #20 (v0.8.0). The two-phase ingest pipeline introduced in
+-- this release runs:
+--    queued → processing(stage=classifying)
+--           → awaiting_review                  (user reviews AI suggestions)
+--           → processing(stage=embedding)      (after user commits)
+--           → done | failed
+--
+-- New columns:
+--   • status gains 'awaiting_review' value
+--   • processing_stage:  'classifying' | 'embedding' | 'upserting' (audit)
+--   • ai_suggestions:    JSONB written by the classifier; consumed by
+--                        the review screen as default form values
+--   • batch_id:          uuid string assigned by the upload route to all
+--                        files in one drag-drop drop. Lets the review
+--                        screen group N files even after a tab refresh.
+--
+-- Idempotent — wrapped in a single transaction.
+--
+-- Run as postgres superuser:
+--   docker exec -i <pg-container> psql -U postgres -d insurance_kb \
+--     -f /tmp/007_ingest_classifier.sql
+
+BEGIN;
+
+ALTER TABLE kb_ingest_job DROP CONSTRAINT IF EXISTS kb_ingest_job_status_check;
+ALTER TABLE kb_ingest_job
+  ADD CONSTRAINT kb_ingest_job_status_check
+  CHECK (status IN ('queued','processing','awaiting_review','done','failed'));
+
+ALTER TABLE kb_ingest_job
+  ADD COLUMN IF NOT EXISTS processing_stage TEXT,
+  ADD COLUMN IF NOT EXISTS ai_suggestions   JSONB,
+  ADD COLUMN IF NOT EXISTS batch_id         TEXT;
+
+CREATE INDEX IF NOT EXISTS kb_ingest_job_batch_idx ON kb_ingest_job (batch_id);
+-- Partial index keeps it tiny: production should have very few rows in
+-- 'awaiting_review' at any moment (it's a transient state).
+CREATE INDEX IF NOT EXISTS kb_ingest_job_awaiting_idx
+  ON kb_ingest_job (created_at DESC) WHERE status = 'awaiting_review';
+
+COMMIT;