feat(kb): admin source management endpoints

Backs the Phase 3 sources table in the KnowledgeBase EspoCRM extension
("ניהול" tab). Four new admin endpoints:

  GET    /admin/kb/sources              — list with chunk_count and
                                          last-ingest status (LATERAL
                                          join against kb_ingest_job),
                                          filterable by topic_id + kind
  PUT    /admin/kb/sources/{id}         — patch the user-editable
                                          metadata only (title,
                                          identifier, source_url,
                                          published_at, effective_at);
                                          structural fields (kind,
                                          topic_id, checksum) are not
                                          accepted
  DELETE /admin/kb/sources/{id}         — hard delete (kb_chunk cascades
                                          via FK; the S3 object at
                                          original_path is best-effort
                                          deleted)
  POST   /admin/kb/sources/{id}/reingest — queues a kb_ingest_job tied
                                           to the existing source_id;
                                           background task replaces
                                           chunks in-place, preserves
                                           the source row and its
                                           hand-edited metadata, and
                                           updates the checksum

The reingest path differs from ingest_source: ingest_source creates a
new kb_source row and supersedes the previous version. Reingest keeps
the same id (so cached chunk_index references in the browser stay
valid), wipes its kb_chunk rows in a single transaction, and inserts
fresh ones from the same file.

Refs Task Master #14 (espocrm-extensions/KnowledgeBase)

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

api/routes/admin_kb.py

@@ -8,7 +8,9 @@ import os
 from typing import Literal
 
 from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
+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 s3 as kb_s3
@@ -316,3 +318,72 @@ async def get_ingest_job(request: Request, job_id: int):
     if not job:
         raise HTTPException(status_code=404, detail="Job not found")
     return job
+
+
+# ── Phase 3: source management (Task #14) ──────────────────────────────────
+
+class SourceUpdate(BaseModel):
+    """Whitelist of editable fields. Anything outside this is rejected."""
+    title: str | None = Field(None, 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
+
+
+@router.get("/sources")
+async def list_admin_sources(
+    request: Request,
+    topic_id: int | None = None,
+    kind: str | None = None,
+    limit: int = 200,
+):
+    _verify_admin(request)
+    if kind and kind not in ("law", "regulation", "circular", "caselaw"):
+        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
+    )
+    return {"items": items, "count": len(items)}
+
+
+@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)
+    try:
+        updated = await kb_admin_sources.update_source(source_id, fields)
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Source not found")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    return updated
+
+
+@router.delete("/sources/{source_id}")
+async def delete_admin_source(request: Request, source_id: int):
+    _verify_admin(request)
+    try:
+        result = await kb_admin_sources.delete_source(source_id)
+    except LookupError:
+        raise HTTPException(status_code=404, detail="Source not found")
+    return result
+
+
+@router.post("/sources/{source_id}/reingest")
+async def reingest_admin_source(request: Request, source_id: int):
+    _verify_admin(request)
+    requested_by = (
+        request.headers.get("X-User-Name")
+        or request.headers.get("X-Forwarded-User")
+        or None
+    )
+    try:
+        job_id = 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}

api/services/kb/admin_sources.py

@@ -0,0 +1,431 @@
+"""Source management for the KB admin panel (Phase 3 / Task #14).
+
+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)
+
+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.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import json
+import logging
+import re
+from typing import Optional
+
+from api.services.kb import ingest as kb_ingest
+from api.services.kb import s3 as kb_s3
+from api.services.kb import voyage
+from api.services.kb.chunker import chunk as chunk_text
+from api.services.kb.db import get_pool
+
+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")
+
+
+# ── Listing ─────────────────────────────────────────────────────────────────
+
+async def list_sources(
+    *,
+    topic_id: Optional[int] = None,
+    kind: Optional[str] = None,
+    limit: int = 200,
+) -> list[dict]:
+    """Source rows enriched with chunk_count and last_ingest_status.
+
+    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.
+    """
+    limit = max(1, min(500, limit))
+    where: list[str] = ["s.superseded_by IS NULL"]
+    args: list = []
+    if topic_id is not None:
+        args.append(topic_id)
+        where.append(f"s.topic_id = ${len(args)}")
+    if kind:
+        args.append(kind)
+        where.append(f"s.kind = ${len(args)}")
+    args.append(limit)
+    sql = f"""
+        SELECT
+            s.id,
+            s.kind,
+            s.title,
+            s.identifier,
+            s.published_at,
+            s.effective_at,
+            s.source_url,
+            s.original_path,
+            s.checksum,
+            s.topic_id,
+            s.created_at,
+            s.updated_at,
+            (SELECT COUNT(*) FROM kb_chunk c WHERE c.source_id = s.id) AS chunk_count,
+            (
+                SELECT json_build_object(
+                    'job_id', j.id,
+                    'status', j.status,
+                    'created_at', j.created_at,
+                    'completed_at', j.completed_at,
+                    'error_message', j.error_message
+                )
+                FROM kb_ingest_job j
+                WHERE j.source_id = s.id
+                ORDER BY j.created_at DESC
+                LIMIT 1
+            ) AS last_ingest
+        FROM kb_source s
+        WHERE {' AND '.join(where)}
+        ORDER BY s.kind, s.title
+        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_source(source_id: int) -> Optional[dict]:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            "SELECT * FROM kb_source WHERE id = $1",
+            source_id,
+        )
+    return _row_to_dict(row) if row else None
+
+
+# ── Editing ─────────────────────────────────────────────────────────────────
+
+async def update_source(source_id: int, fields: dict) -> dict:
+    """UPDATE only whitelisted columns. Returns the post-update row."""
+    clean: dict = {}
+    for k in _EDITABLE_FIELDS:
+        if k not in fields:
+            continue
+        v = fields[k]
+        if k in ("published_at", "effective_at"):
+            clean[k] = kb_ingest._coerce_date(v)
+        elif v is None or v == "":
+            clean[k] = None
+        else:
+            clean[k] = str(v).strip()
+    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 not row:
+        raise LookupError(f"source {source_id} not found")
+    return _row_to_dict(row)
+
+
+# ── Deletion ────────────────────────────────────────────────────────────────
+
+async def delete_source(source_id: int) -> dict:
+    """Hard-delete a source. kb_chunk rows cascade via FK; the underlying
+    S3 object is deleted on a best-effort basis (a stray file is harmless,
+    but cluttered).
+    """
+    src = await get_source(source_id)
+    if not src:
+        raise LookupError(f"source {source_id} not found")
+
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        result = await conn.execute(
+            "DELETE FROM kb_source WHERE id = $1", source_id
+        )
+    deleted = result.endswith(" 1")
+
+    # Best-effort S3 cleanup — never fail the whole delete on a stray
+    # object that might already be gone.
+    s3_deleted = False
+    bucket, key = _parse_s3_uri(src.get("original_path") or "")
+    if key:
+        try:
+            kb_s3._client().delete_object(Bucket=bucket or kb_s3._bucket(), Key=key)
+            s3_deleted = True
+        except Exception as exc:
+            logger.warning("[admin.delete] S3 delete failed for %s: %s", key, exc)
+
+    return {"deleted": deleted, "source_id": source_id, "s3_deleted": s3_deleted}
+
+
+# ── Re-ingest ───────────────────────────────────────────────────────────────
+
+async def start_reingest(source_id: int, requested_by_user: Optional[str]) -> int:
+    """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.
+    """
+    src = await get_source(source_id)
+    if not src:
+        raise LookupError(f"source {source_id} not found")
+    if not src.get("original_path"):
+        raise ValueError(
+            f"source {source_id} has no original_path — re-ingest needs the original file"
+        )
+
+    # Pre-flight: confirm the file actually exists in S3 before queuing
+    # the job. Saves the user from waiting 30s on a doomed job.
+    bucket, key = _parse_s3_uri(src["original_path"])
+    if not key:
+        raise ValueError(f"unparseable original_path: {src['original_path']}")
+    try:
+        kb_s3._client().head_object(Bucket=bucket or kb_s3._bucket(), Key=key)
+    except Exception as exc:
+        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).
+    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"),
+        "reingest": True,
+    }
+    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)
+            RETURNING id
+            """,
+            source_id,
+            src.get("title") or _filename_from_key(key),
+            key,
+            src["kind"],
+            src["topic_id"],
+            json.dumps(metadata, ensure_ascii=False),
+            requested_by_user,
+        )
+    job_id = row["id"]
+    logger.info("[admin.reingest] queued job_id=%s source_id=%s", job_id, source_id)
+    return job_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.
+
+    Errors are written to the job row (status='failed', error_message=...)
+    so the polling UI never gets stuck on 'processing'.
+    """
+    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,
+        )
+    if not job_row:
+        logger.warning("[reingest] job %s not in queued state, skipping", job_id)
+        return
+
+    job = _row_to_dict(job_row)
+    source_id = job["source_id"]
+    if source_id is None:
+        await _fail_job(pool, job_id, "re-ingest job has no source_id")
+        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.
+        if not key:
+            key = job["s3_key"]
+        data = kb_s3._client().get_object(
+            Bucket=bucket or kb_s3._bucket(),
+            Key=key,
+        )["Body"].read()
+        if not data:
+            raise kb_ingest.IngestError(f"empty file at {key}")
+
+        text = kb_ingest.parse(data, job.get("original_filename") or key)
+        if not text:
+            raise kb_ingest.IngestError("no text extracted")
+
+        chunks = chunk_text(text, job["kind"])
+        if not chunks:
+            raise kb_ingest.IngestError("chunker produced zero chunks")
+
+        # Embed first (slow, network-bound) so the actual DB swap is fast.
+        texts = [c["content"] for c in chunks]
+        vectors = await voyage.embed(texts, input_type="document")
+
+        import hashlib
+        new_checksum = hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+        async with pool.acquire() as conn:
+            async with conn.transaction():
+                # Drop existing chunks and reinsert in one tx so we never
+                # leave the source partially-chunked.
+                await conn.execute(
+                    "DELETE FROM kb_chunk WHERE source_id = $1", source_id
+                )
+                rows = [
+                    (
+                        source_id,
+                        c["chunk_index"],
+                        c["heading_path"],
+                        c["section_ref"],
+                        c["content"],
+                        vectors[i],
+                        c["token_count"],
+                        c.get("page_number"),
+                    )
+                    for i, c in enumerate(chunks)
+                ]
+                await conn.executemany(
+                    """
+                    INSERT INTO kb_chunk
+                        (source_id, chunk_index, heading_path, section_ref,
+                         content, embedding, token_count, page_number)
+                    VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
+                    """,
+                    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,
+                )
+                await conn.execute(
+                    """
+                    UPDATE kb_ingest_job
+                       SET status = 'done',
+                           chunks_created = $2,
+                           completed_at = now()
+                     WHERE id = $1
+                    """,
+                    job_id, len(chunks),
+                )
+        logger.info(
+            "[reingest] done job_id=%s source_id=%s chunks=%d",
+            job_id, source_id, len(chunks),
+        )
+    except Exception as exc:
+        logger.exception("[reingest] failed job_id=%s", job_id)
+        await _fail_job(pool, job_id, str(exc))
+
+
+# ── helpers ─────────────────────────────────────────────────────────────────
+
+async def _fail_job(pool, job_id: int, msg: str) -> None:
+    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, msg[:4000],
+        )
+
+
+def _row_to_dict(row) -> Optional[dict]:
+    if row is None:
+        return None
+    out: dict = {}
+    for k, v in dict(row).items():
+        if isinstance(v, _dt.datetime):
+            out[k] = v.isoformat()
+        elif isinstance(v, _dt.date):
+            out[k] = v.isoformat()
+        elif k == "metadata_json" and isinstance(v, str):
+            try:
+                out[k] = json.loads(v)
+            except Exception:
+                out[k] = {}
+        elif k == "last_ingest" and isinstance(v, str):
+            try:
+                out[k] = json.loads(v)
+            except Exception:
+                out[k] = None
+        else:
+            out[k] = v
+    return out
+
+
+_S3_URI = re.compile(r"^s3://([^/]+)/(.+)$")
+
+def _parse_s3_uri(uri: str) -> tuple[str, str]:
+    """('bucket', 'key/path') or ('', '') if not an s3:// URI."""
+    if not uri:
+        return ("", "")
+    m = _S3_URI.match(uri)
+    if not m:
+        return ("", "")
+    return (m.group(1), m.group(2))
+
+
+def _filename_from_key(key: str) -> str:
+    return key.rsplit("/", 1)[-1] if key else "source"
+
+
+def _date_to_iso(value) -> Optional[str]:
+    if value is None:
+        return None
+    if isinstance(value, str):
+        return value[:10] if value else None
+    if isinstance(value, _dt.date):
+        return value.isoformat()
+    return None