brain-of-reese/app/api/sync.py

"""Sources sync API — one-click KB mirror (phase 32, task 01).

Admin-only ``POST /api/sync`` + ``GET /api/sync/status`` behind the
existing :func:`app.core.auth.require_admin` (A10 extended, phase 16
pattern — the public API surface stays stateless, the signed cookie
remains the only session state, same as ``/api/steering``).

The button's backend runs the full document sync **in-process** (A12
untouched — no queue, no new services): one ``asyncio`` background task
plus a module-level :class:`SyncStatus` that the UI polls every 2 s
(task 02). One sync at a time — ``POST`` while a run is in flight is
409; the status object is authoritative, so the UI can never sit on a
stale button state (§7.4 adaptation, phase locked decisions).

Pipeline (the canonical "mirror the sources" action — phase locked
decisions):

1. verify ``embed`` + summary model availability — fail fast before
   any clone (:func:`app.rag.llm.check_models`, phase 41): a dead
   model endpoint aborts the run naming the unavailable model, before
   source resolution or any ``clone_or_pull``;
2. resolve the effective sources — the ``git_sources`` DB rows (git
   **and** local, phase 38), else the ``BOR_GIT_SOURCES`` fallback
   (git-only)
   (:func:`app.rag.git_sources.effective_sources`, shared with the
   CLI) — empty on both origins (no git rows, no local rows, no env
   URLs) fails loudly (``no sources configured (git or local)``)
   instead of silently importing the legacy local directories;
3. per resolved row: ``kind=git`` → :func:`scripts.git_sync.clone_or_pull`
   into ``BOR_SOURCES_DIR/<repo-name>/`` (phase 28 — reused, not
   re-implemented); ``kind=local`` → the stored directory, re-verified
   ``.is_dir()`` **at sync time** (it may have moved/deleted since
   add-time) — a missing directory raises ``local source missing:
   <path>``; a failing clone or a missing local dir aborts before any
   import;
4. ``import_sources(..., prune=True)`` over the single combined list
   (git checkouts + local dirs) — prune so files deleted upstream or
   out of a local dir leave the index (pruning covers the union; the
   CLI's no-prune default is unchanged);
5. when the import changed the KB (added + updated > 0),
   ``regenerate_overview`` refreshes the single ``kb_overview`` row
   (phase 31 trigger, best-effort inside);
6. when the import changed the KB (added + updated + pruned > 0 — the
   saved-chat invalidation gate, phase 53 task 02: a pruned document
   can invalidate a saved answer that cited it, deliberately broader
   than step 5's overview gate), the single-row ``sources_meta``
   version counter is bumped exactly once in a short-lived session and
   the resulting generation lands in the status detail as
   ``sources_version`` (an unchanged re-sync reports the current
   generation without advancing it). A FAILED sync never bumps — the
   run aborts in the ``failed`` state before this step.

Status is in memory: a restart mid-sync loses the running state
(accepted — the next click re-syncs idempotently). The status also
carries the phase-64 per-file progress — ``current_file`` (the
``source/relative/path`` the import is processing right now) plus
``files_done`` / ``files_total`` — null/0/0 before the import starts
(clone/pull reports no file yet) and in terminal states, which clear
``current_file`` but keep the run's final counts.

The ``failed`` state's ``error`` string is masked by the shared
sanitizer — the ``user:pass@`` masker now lives in :mod:`app.core.errors`
(imported here under the private name ``_sanitize_error``).
"""
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Literal

from fastapi import APIRouter, Depends, HTTPException

from app.config import get_settings
from app.core.auth import require_admin
from app.core.errors import sanitize_error as _sanitize_error
from app.db import SessionLocal
from app.rag.git_sources import effective_sources
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient, check_models
from app.rag.overview import regenerate_overview
from app.rag.sources_meta import bump_sources_version, current_sources_version
from scripts.git_sync import GitSyncError, clone_or_pull
from scripts.import_docs import repo_name

logger = logging.getLogger("app.api.sync")

router = APIRouter(
    prefix="/sync",
    tags=["sync"],
    dependencies=[Depends(require_admin)],  # phase 16 pattern: admin-only surface
)


@dataclass
class SyncStatus:
    """In-memory state of the (at most one) in-flight sync run.

    ``state`` is a four-state machine: ``idle`` (never run / reset),
    ``running``, ``success``, ``failed``. Terminal states carry the run's
    ``detail`` (success) or ``error`` (failure) so the UI can render the
    last result after a page reload (task 02's re-attach behavior).

    Phase 64 (task 02) progress fields: ``current_file`` is the
    ``source/relative/path`` the import is processing right now (null
    outside the import phase — clone/pull first, terminal states
    after); ``files_done`` / ``files_total`` carry the hook's
    done/total position and survive a terminal state (the run's last
    position is useful context next to the error).
    """

    state: Literal["idle", "running", "success", "failed"] = "idle"
    started_at: datetime | None = None
    finished_at: datetime | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    error: str | None = None
    # Phase 64 (task 02): per-file progress — the file the import is
    # processing right now and the hook's done/total position.
    current_file: str | None = None
    files_done: int = 0
    files_total: int = 0


_status = SyncStatus()
_task: asyncio.Task[None] | None = None


@router.get("/status")
def sync_status() -> dict[str, Any]:
    """Current sync state (the UI polls this every 2 s — task 02).

    ``started_at`` / ``finished_at`` are ISO-8601 strings or null.
    ``current_file`` (phase 64) is the ``source/relative/path`` the
    import is processing right now — null during the clone/pull phase
    and in terminal states; ``files_done`` / ``files_total`` carry the
    hook's position (0/0 idle).
    """
    return {
        "state": _status.state,
        "started_at": _status.started_at.isoformat() if _status.started_at else None,
        "finished_at": _status.finished_at.isoformat() if _status.finished_at else None,
        "detail": _status.detail,
        "error": _status.error,
        "current_file": _status.current_file,
        "files_done": _status.files_done,
        "files_total": _status.files_total,
    }


@router.post("", status_code=202)
async def start_sync() -> dict[str, str]:
    """Start the clone → import → overview sync as a background task.

    202 + ``sync started`` kicks off :func:`_run_sync` on the app's event
    loop. 409 when a run is already in flight (one sync at a time — the
    status endpoint is the single source of truth for the run, and the
    UI re-attaches to it rather than starting a second one).
    """
    global _task
    if _task is not None and not _task.done():
        raise HTTPException(status_code=409, detail="a sync is already running")
    _task = asyncio.create_task(_run_sync())
    return {"detail": "sync started"}


async def _run_sync() -> None:
    """The full sync pipeline, one in-process background task.

    Every failure mode (git, embeddings, anything else) lands in the
    ``failed`` state with a sanitized ``error`` string — a background
    task must die in state, never as an unobserved exception.
    ``CancelledError`` is deliberately *not* caught: app shutdown
    cancels the task, and swallowing that would mask a real stop.
    """
    _status.state = "running"
    _status.started_at = datetime.now(UTC)
    _status.finished_at = None
    _status.detail = {}
    _status.error = None
    # Phase 64 (task 02): the progress fields reset with the run — no
    # current file until the import starts (the clone/pull phase).
    _status.current_file = None
    _status.files_done = 0
    _status.files_total = 0
    try:
        settings = get_settings()
        # Step 1 (phase 41): fail fast — verify both models the sync
        # needs (embed + summary) before source resolution or any
        # clone. The client is reused for the import + overview below.
        llm = LLMClient()
        await check_models(llm)
        # The background task has no request session: open a short-lived
        # one around the shared phase-35/38 resolver (DB rows of both
        # kinds win; the BOR_GIT_SOURCES git list is a fallback while
        # the table is empty).
        db = SessionLocal()
        try:
            rows, origin = effective_sources(db)
        finally:
            db.close()
        if not rows:
            # The button targets the admin-managed source registry
            # (manual --source dirs have no repo to clone) — an empty
            # config on *both* origins (no git rows, no local rows, no
            # env URLs) fails loudly instead of silently importing the
            # legacy directories.
            raise GitSyncError("no sources configured (git or local)")
        git_count = sum(1 for row in rows if row.kind == "git")
        logger.info(
            "sync: started repos=%d origin=%s git=%d local=%d",
            len(rows), origin, git_count, len(rows) - git_count,
        )
        sources_root = Path(settings.sources_dir).expanduser()
        sources: list[Path] = []
        for row in rows:
            if row.kind == "git":
                sources.append(clone_or_pull(row.url, sources_root / repo_name(row.url)))
            else:
                # kind=local — the stored expanded path (phase 38 also
                # mirrors it in the NOT-NULL ``url`` location column, the
                # ``or`` keeps the type checker honest); re-verified at
                # sync time because the directory may have moved or been
                # deleted since add-time.
                path = Path(row.path or row.url).expanduser()
                if not path.is_dir():
                    raise GitSyncError(f"local source missing: {path}")
                sources.append(path)
        # Phase 64 (task 02): the per-file progress hook — the status
        # endpoint reports the file being processed right now. The
        # closure captures the module ``_status`` exactly like the state
        # assignments above.
        def _hook(source: str, rel: str, done: int, total: int) -> None:
            _status.current_file = f"{source}/{rel}"
            _status.files_done = done
            _status.files_total = total

        summary: ImportSummary = await import_sources(sources, llm, prune=True, progress=_hook)
        overview = False
        if summary.added + summary.updated > 0:
            overview = await regenerate_overview(llm)
        # Phase 53 (task 02): a sync that changed the KB advances the
        # sources version exactly once — the saved-chat invalidation
        # marker (task 03 stamps rows against it). The gate is
        # deliberately broader than the overview's above: a pruned
        # document can invalidate a saved answer that cited it, so
        # ``pruned > 0`` bumps too. The bump commits in its own short
        # session (the ``effective_sources`` pattern above), so it
        # lands even if the best-effort overview then fails — the index
        # really did change. An unchanged re-sync never bumps; it
        # reports the current generation instead, so the detail always
        # carries the generation the KB is now at.
        db = SessionLocal()
        try:
            if summary.added + summary.updated + summary.pruned > 0:
                sources_version = bump_sources_version(db)
                db.commit()
            else:
                sources_version = current_sources_version(db)
        finally:
            db.close()
        _status.state = "success"
        _status.finished_at = datetime.now(UTC)
        _status.current_file = None  # phase 64: keep the final counts
        _status.detail = {
            "files": summary.files,
            "added": summary.added,
            "updated": summary.updated,
            "unchanged": summary.unchanged,
            "pruned": summary.pruned,
            "errors": summary.errors,
            "chunks": summary.chunks,
            "summaries": summary.summaries,
            "summary_errors": summary.summary_errors,
            "overview": overview,
            "sources_version": sources_version,
        }
        logger.info("sync: done detail=%s", _status.detail)
    except Exception as e:  # noqa: BLE001 — a background task dies in state, see above
        logger.exception("sync: failed")
        _status.state = "failed"
        _status.finished_at = datetime.now(UTC)
        _status.error = _sanitize_error(str(e))
        _status.current_file = None  # phase 64: keep the final counts