diff --git a/api/routes/admin_kb.py b/api/routes/admin_kb.py index 1ab8574..6076698 100644 --- a/api/routes/admin_kb.py +++ b/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)) - return {"job_id": job_id, "status": "queued", "source_id": source_id} + # Schedule classify (fail-soft) — same path as fresh uploads, so + # 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) ───────────────────────────────────────── diff --git a/api/services/kb/admin_sources.py b/api/services/kb/admin_sources.py index 18e3699..d088318 100644 --- a/api/services/kb/admin_sources.py +++ b/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 - update_source — patch the user-editable metadata - (title, identifier, dates, source_url) - delete_source — hard delete (cascades chunks; drops S3 object) - start_reingest — re-parse the existing file in place; preserves - source_id and metadata, replaces chunks - process_reingest_job — async background worker that does the actual - parse→chunk→embed and swaps chunks in a single - transaction (UI polls the same /admin/kb/jobs - endpoint as upload jobs) + list_sources — table view with chunk count and last-ingest status + update_source — patch the user-editable metadata + (title, identifier, dates, source_url) + delete_source — hard delete (cascades chunks; drops S3 object) + start_reingest — kick off the two-phase re-ingest pipeline: + classify → awaiting_review → commit. The + classify stage reuses + ingest_jobs.process_classify_stage so the + user gets fresh AI metadata suggestions in + 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 - `asyncio.create_task` from the route. Returns the new job_id so the - browser can poll using the same /admin/kb/jobs/{id} endpoint as - upload jobs. + The job enters the same two-phase pipeline as a fresh upload — + classify (AI metadata extraction) → awaiting_review → embed + (re-chunk + UPDATE kb_source) — so the user can review suggestions + 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/` 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) - VALUES ($1, $2, $3, $4, $5, $6::jsonb, 'queued', $7) + metadata_json, status, requested_by_user, batch_id) + 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) - return job_id + logger.info( + "[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: - """Background worker for re-ingest. Replaces chunks in-place; preserves - the kb_source row and its metadata. +async def process_reingest_embed_stage( + job_id: int, fields: dict, label_ids: list[int], +) -> None: + """Final stage of the two-phase re-ingest pipeline. - Errors are written to the job row (status='failed', error_message=...) - so the polling UI never gets stuck on 'processing'. + Called from `ingest_jobs.commit_job` after the user reviewed the AI + 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( - """ - UPDATE kb_ingest_job - SET status = 'processing', started_at = now() - WHERE id = $1 AND status = 'queued' - RETURNING * - """, - job_id, + "SELECT * FROM kb_ingest_job WHERE id = $1", 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 - # If s3_key is a bare key (no s3:// prefix), _parse_s3_uri returns - # ("", "") — fall back to the bare key. + # Step 1: UPDATE kb_source with the user-confirmed metadata. + # commit_job's `fields` dict mirrors JobCommit: kind, title, + # 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", - job_id, source_id, len(chunks), + "[reingest-embed] done job_id=%s source_id=%s chunks=%d labels=%d", + 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: diff --git a/api/services/kb/ingest_jobs.py b/api/services/kb/ingest_jobs.py index 10b7167..1688981 100644 --- a/api/services/kb/ingest_jobs.py +++ b/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}