feat(kb-reingest): re-classify metadata via AI on every re-process

Old behavior: clicking "עיבוד מחדש" on a source preserved the existing
metadata (title, identifier, dates, summary, labels) and only re-chunked
+ re-embedded the file. Sources whose original AI classification produced
weak metadata (e.g. title="החלטה" with no summary) had no path to recover
short of deleting and re-uploading.

New behavior: re-process now goes through the same two-phase pipeline as
a fresh upload — classify (AI extracts metadata) → awaiting_review (user
sees suggestions in the batch review screen and edits) → commit (kb_source
metadata is overwritten + chunks are replaced + labels are replaced).

Implementation:
  • start_reingest now returns {job_id, batch_id} and pre-fills metadata_json
    with the existing source's labels + summary so the form has a fallback
    if the classifier times out.
  • The reingest route schedules ingest_jobs.process_classify_stage instead
    of admin_sources.process_reingest_job.
  • commit_job branches on job.source_id: NULL → fresh upload (creates new
    kb_source); set → re-ingest (UPDATE kb_source + replace chunks).
  • New process_reingest_embed_stage replaces the old process_reingest_job;
    it bypasses _EDITABLE_FIELDS so the user can also change kind on
    re-process (e.g. caselaw uploaded as circular).

The browser opens the same batch review screen, so the UX matches uploads
exactly. Re-ingests are single-file batches.

Frontend changes (KnowledgeBase extension v0.9.0) ship in a separate repo.

api/routes/admin_kb.py

@@ -380,6 +380,13 @@ async def delete_admin_source(request: Request, source_id: int):
 
 @router.post("/sources/{source_id}/reingest")
 async def reingest_admin_source(request: Request, source_id: int):
+    """Kick off a two-phase re-ingest: AI re-classifies the existing
+    file, the user reviews suggestions on the batch review screen, and
+    the commit overwrites kb_source metadata + replaces chunks.
+
+    Returns `batch_id` so the browser can open the same review UI as
+    a fresh upload (single-file batch).
+    """
     _verify_admin(request)
     requested_by = (
         request.headers.get("X-User-Name")
@@ -387,14 +394,21 @@ async def reingest_admin_source(request: Request, source_id: int):
         or None
     )
     try:
-        job_id = await kb_admin_sources.start_reingest(source_id, requested_by)
+        result = await kb_admin_sources.start_reingest(source_id, requested_by)
     except LookupError:
         raise HTTPException(status_code=404, detail="Source not found")
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
 
-    asyncio.create_task(kb_admin_sources.process_reingest_job(job_id))
+    # Schedule classify (fail-soft) — same path as fresh uploads, so
-    return {"job_id": job_id, "status": "queued", "source_id": source_id}
+    # the user gets fresh AI metadata suggestions on the review screen.
+    asyncio.create_task(kb_jobs.process_classify_stage(result["job_id"]))
+    return {
+        "job_id": result["job_id"],
+        "batch_id": result["batch_id"],
+        "status": "queued",
+        "source_id": source_id,
+    }
 
 
 # ── Phase 4: topic CRUD (Task #15) ─────────────────────────────────────────

api/services/kb/admin_sources.py

@@ -4,22 +4,31 @@ Today the only way to clean up or correct a source is `DELETE FROM
 kb_source WHERE id=...` on the shared PG plus an mc rm in MinIO. This
 module backs the new sources table in the EspoCRM "ניהול" tab:
 
-  list_sources       — table view with chunk count and last-ingest status
+  list_sources              — table view with chunk count and last-ingest status
-  update_source      — patch the user-editable metadata
+  update_source             — patch the user-editable metadata
-                       (title, identifier, dates, source_url)
+                              (title, identifier, dates, source_url)
-  delete_source      — hard delete (cascades chunks; drops S3 object)
+  delete_source             — hard delete (cascades chunks; drops S3 object)
-  start_reingest     — re-parse the existing file in place; preserves
+  start_reingest            — kick off the two-phase re-ingest pipeline:
-                       source_id and metadata, replaces chunks
+                              classify → awaiting_review → commit. The
-  process_reingest_job — async background worker that does the actual
+                              classify stage reuses
-                         parse→chunk→embed and swaps chunks in a single
+                              ingest_jobs.process_classify_stage so the
-                         transaction (UI polls the same /admin/kb/jobs
+                              user gets fresh AI metadata suggestions in
-                         endpoint as upload jobs)
+                              the same review UI as fresh uploads.
+  process_reingest_embed_stage — final stage after the user commits:
+                              UPDATE kb_source metadata + replace labels +
+                              re-chunk and re-embed (the original
+                              process_reingest_job logic, minus the
+                              "preserve metadata" rule).
 
 The reingest path differs from ingest_source: ingest_source creates a
 NEW kb_source row and supersedes the previous version. Reingest keeps
 the same row id (so cached _lastSearch chunk_index references in the
 browser still point at a valid source), wipes its chunks, and inserts
-fresh ones from the same file. Hand-edited metadata is preserved.
+fresh ones from the same file. As of v0.9.0, hand-edited metadata is
+NO LONGER preserved silently — the user reviews AI suggestions and
+edits them on the batch review screen before committing, so any
+overwrite is explicit. Sources whose AI-extracted metadata was thin
+(e.g. title="החלטה" only) get fresh suggestions on every re-process.
 """
 
 from __future__ import annotations
@@ -244,14 +253,22 @@ async def delete_source(source_id: int) -> dict:
 
 # ── Re-ingest ───────────────────────────────────────────────────────────────
 
-async def start_reingest(source_id: int, requested_by_user: Optional[str]) -> int:
+async def start_reingest(
+    source_id: int, requested_by_user: Optional[str]
+) -> dict:
     """Create a kb_ingest_job tied to the existing source_id.
 
-    The actual work is in `process_reingest_job`, scheduled via
+    The job enters the same two-phase pipeline as a fresh upload —
-    `asyncio.create_task` from the route. Returns the new job_id so the
+    classify (AI metadata extraction) → awaiting_review → embed
-    browser can poll using the same /admin/kb/jobs/{id} endpoint as
+    (re-chunk + UPDATE kb_source) — so the user can review suggestions
-    upload jobs.
+    on the batch review screen before anything overwrites kb_source.
+
+    Returns {"job_id", "batch_id"}; the route schedules
+    `ingest_jobs.process_classify_stage(job_id)` after this, and the
+    browser opens `/admin/kb/batch/<batch_id>` to poll the review UI.
     """
+    import uuid as _uuid
+
     src = await get_source(source_id)
     if not src:
         raise LookupError(f"source {source_id} not found")
@@ -271,23 +288,33 @@ async def start_reingest(source_id: int, requested_by_user: Optional[str]) -> in
         raise ValueError(f"original file unavailable: {exc}")
 
     # Reuse the kb_ingest_job table; mark this row as a re-ingest by
-    # pre-filling source_id (uploads start with source_id NULL).
+    # pre-filling source_id (uploads start with source_id NULL). The
+    # batch_id is auto-generated so the existing review UI works
+    # without changes — re-ingests are single-file batches. Existing
+    # labels go into metadata_json so the review form can pre-fill
+    # them as a fallback when the AI classifier hasn't run yet (or
+    # returns empty).
+    from api.services.kb import labels as kb_labels  # local import: cycle
+    existing_labels = await kb_labels.get_labels_for_source(source_id)
     metadata = {
         "title": src.get("title"),
         "identifier": src.get("identifier"),
         "published_at": _date_to_iso(src.get("published_at")),
         "effective_at": _date_to_iso(src.get("effective_at")),
         "source_url": src.get("source_url"),
+        "summary": src.get("description"),
+        "labels": [lbl.get("name") for lbl in existing_labels if lbl.get("name")],
         "reingest": True,
     }
+    batch_id = _uuid.uuid4().hex
     pool = await get_pool()
     async with pool.acquire() as conn:
         row = await conn.fetchrow(
             """
             INSERT INTO kb_ingest_job
                 (source_id, original_filename, s3_key, kind, topic_id,
-                 metadata_json, status, requested_by_user)
+                 metadata_json, status, requested_by_user, batch_id)
-            VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'queued', $7)
+            VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'queued', $7, $8)
             RETURNING id
             """,
             source_id,
@@ -297,32 +324,46 @@ async def start_reingest(source_id: int, requested_by_user: Optional[str]) -> in
             src["topic_id"],
             json.dumps(metadata, ensure_ascii=False),
             requested_by_user,
+            batch_id,
         )
     job_id = row["id"]
-    logger.info("[admin.reingest] queued job_id=%s source_id=%s", job_id, source_id)
+    logger.info(
-    return job_id
+        "[admin.reingest] queued job_id=%s source_id=%s batch_id=%s",
+        job_id, source_id, batch_id,
+    )
+    return {"job_id": job_id, "batch_id": batch_id}
 
 
-async def process_reingest_job(job_id: int) -> None:
+async def process_reingest_embed_stage(
-    """Background worker for re-ingest. Replaces chunks in-place; preserves
+    job_id: int, fields: dict, label_ids: list[int],
-    the kb_source row and its metadata.
+) -> None:
+    """Final stage of the two-phase re-ingest pipeline.
 
-    Errors are written to the job row (status='failed', error_message=...)
+    Called from `ingest_jobs.commit_job` after the user reviewed the AI
-    so the polling UI never gets stuck on 'processing'.
+    suggestions. The job is already in `status='processing',
+    processing_stage='embedding'` at this point — commit_job set that
+    transition before scheduling us.
+
+    Steps:
+      1. UPDATE kb_source with the user-confirmed metadata (kind,
+         title, identifier, source_url, dates, description=summary).
+      2. Replace the source's labels with `label_ids`.
+      3. Re-parse the file → chunk → embed → atomic DELETE+INSERT
+         of kb_chunk rows + checksum update.
+      4. Mark the job done.
+
+    Failures here are real failures (the user already confirmed). They
+    surface as `status='failed'` so the polling UI lets the user retry.
     """
+    from api.services.kb import labels as kb_labels  # local import: cycle
+
     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,
-            UPDATE kb_ingest_job
-               SET status = 'processing', started_at = now()
-             WHERE id = $1 AND status = 'queued'
-            RETURNING *
-            """,
-            job_id,
         )
     if not job_row:
-        logger.warning("[reingest] job %s not in queued state, skipping", job_id)
+        logger.warning("[reingest-embed] job %s vanished", job_id)
         return
 
     job = _row_to_dict(job_row)
@@ -332,11 +373,34 @@ async def process_reingest_job(job_id: int) -> None:
         return
 
     try:
-        bucket, key = _parse_s3_uri(job["s3_key"])  # s3_key here is just the key
+        # Step 1: UPDATE kb_source with the user-confirmed metadata.
-        # If s3_key is a bare key (no s3:// prefix), _parse_s3_uri returns
+        # commit_job's `fields` dict mirrors JobCommit: kind, title,
-        # ("", "") — fall back to the bare key.
+        # identifier, source_url, published_at, effective_at, summary.
+        # kb_source.description is the schema name for "summary".
+        kind = (fields.get("kind") or job["kind"]).strip()
+        title = (fields.get("title") or "").strip()
+        if not title:
+            raise ValueError("title is empty after commit — aborting re-ingest")
+
+        await _update_source_metadata_for_reingest(
+            pool=pool,
+            source_id=source_id,
+            kind=kind,
+            title=title,
+            identifier=fields.get("identifier"),
+            source_url=fields.get("source_url"),
+            published_at=fields.get("published_at"),
+            effective_at=fields.get("effective_at"),
+            description=fields.get("summary"),
+        )
+
+        # Step 2: replace labels (empty list clears all).
+        await kb_labels.replace_for_source(source_id, label_ids or [])
+
+        # Step 3: re-chunk + re-embed.
+        bucket, key = _parse_s3_uri(job["s3_key"])
         if not key:
-            key = job["s3_key"]
+            key = job["s3_key"]  # bare key, no s3:// prefix
         data = kb_s3._client().get_object(
             Bucket=bucket or kb_s3._bucket(),
             Key=key,
@@ -348,7 +412,7 @@ async def process_reingest_job(job_id: int) -> None:
         if not text:
             raise kb_ingest.IngestError("no text extracted")
 
-        chunks = chunk_text(text, job["kind"])
+        chunks = chunk_text(text, kind)
         if not chunks:
             raise kb_ingest.IngestError("chunker produced zero chunks")
 
@@ -388,8 +452,6 @@ async def process_reingest_job(job_id: int) -> None:
                     """,
                     rows,
                 )
-                # Update only the structural fields that depend on the file
-                # (checksum). Hand-edited metadata stays as-is.
                 await conn.execute(
                     "UPDATE kb_source SET checksum = $2 WHERE id = $1",
                     source_id, new_checksum,
@@ -398,6 +460,7 @@ async def process_reingest_job(job_id: int) -> None:
                     """
                     UPDATE kb_ingest_job
                        SET status = 'done',
+                           processing_stage = NULL,
                            chunks_created = $2,
                            completed_at = now()
                      WHERE id = $1
@@ -405,14 +468,50 @@ async def process_reingest_job(job_id: int) -> None:
                     job_id, len(chunks),
                 )
         logger.info(
-            "[reingest] done job_id=%s source_id=%s chunks=%d",
+            "[reingest-embed] done job_id=%s source_id=%s chunks=%d labels=%d",
-            job_id, source_id, len(chunks),
+            job_id, source_id, len(chunks), len(label_ids or []),
         )
     except Exception as exc:
-        logger.exception("[reingest] failed job_id=%s", job_id)
+        logger.exception("[reingest-embed] failed job_id=%s", job_id)
         await _fail_job(pool, job_id, str(exc))
 
 
+async def _update_source_metadata_for_reingest(
+    *, pool, source_id: int, kind: str, title: str,
+    identifier: Optional[str], source_url: Optional[str],
+    published_at: Optional[str], effective_at: Optional[str],
+    description: Optional[str],
+) -> None:
+    """UPDATE kb_source after a re-ingest commit. Bypasses _EDITABLE_FIELDS
+    so we can also update `kind` (the user explicitly chose it on the
+    review screen — e.g. caselaw uploaded as circular by mistake).
+    """
+    pub = kb_ingest._coerce_date(published_at) if published_at else None
+    eff = kb_ingest._coerce_date(effective_at) if effective_at else None
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            UPDATE kb_source
+               SET kind = $2,
+                   title = $3,
+                   identifier = $4,
+                   source_url = $5,
+                   published_at = $6,
+                   effective_at = $7,
+                   description = $8
+             WHERE id = $1
+            """,
+            source_id,
+            kind,
+            title,
+            (identifier or None),
+            (source_url or None),
+            pub,
+            eff,
+            (description or None),
+        )
+
+
 # ── helpers ─────────────────────────────────────────────────────────────────
 
 async def _fail_job(pool, job_id: int, msg: str) -> None:

api/services/kb/ingest_jobs.py

@@ -389,6 +389,11 @@ async def commit_job(
     `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.
+
+    Branches on `job.source_id`:
+      • NULL  → fresh upload → `process_embed_stage` (creates kb_source).
+      • set   → re-ingest    → `process_reingest_embed_stage` (UPDATEs
+                              the existing kb_source + replaces chunks).
     """
     pool = await get_pool()
 
@@ -426,7 +431,17 @@ async def commit_job(
             json.dumps(fields, ensure_ascii=False),
         )
 
-    asyncio.create_task(process_embed_stage(job_id, fields, label_ids))
+    if job.get("source_id"):
+        # Re-ingest: UPDATE the existing kb_source + replace chunks.
+        from api.services.kb import admin_sources as kb_admin_sources
+        asyncio.create_task(
+            kb_admin_sources.process_reingest_embed_stage(
+                job_id, fields, label_ids,
+            )
+        )
+    else:
+        # Fresh upload: ingest_source creates a new kb_source row.
+        asyncio.create_task(process_embed_stage(job_id, fields, label_ids))
     return {"job_id": job_id, "status": "processing", "label_ids": label_ids}