"""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). Phase 134 (task 01) adds the cancel: ``POST /api/sync/cancel`` (admin-only, same router) registers a COOPERATIVE cancel while a run is in flight (202, 409 ``no sync is running`` otherwise), and the run lands in the new ``cancelled`` TERMINAL state — a cancel is intent, not failure (``error = None``, ``detail = {}``, the final counts kept; the cancel paragraph below). 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` with the phase-121 clone URL (:func:`app.rag.git_sources.clone_url_for` — the row's ``token`` column injected as ``https://x-access-token:@…`` only for https? rows; NULL token → the bare stored URL verbatim, so public repos and legacy embedded-token rows clone exactly as before) and the row's BRANCH PIN as the third argument (phase 135 — a fresh clone checks it out, ``git clone --branch``; an existing checkout is guarded — a mismatched branch fails loudly, the remove-and-re-add error — and tracked, ``git fetch origin `` + ``git merge --ff-only FETCH_HEAD``; ``branch=None``, i.e. every pre-phase-135 row and every env-fallback row, keeps the pre-phase argv byte-identical) into ``BOR_SOURCES_DIR//`` (phase 28 — reused, not re-implemented; the checkout name stays on the bare URL — credential-free); ``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: ``; 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), honoring each row's ``ignore_paths`` (phase 89 — the per-root ignore map is built in the same per-row loop as the source list), its ``include_hidden`` flag (phase 105 — the per-root hidden-folders map, same per-row construction), and feeding each git checkout's per-file last-commit dates (phase 106, D2 — the per-root date map, ``file_commit_dates`` after the clone, same per-row construction; local rows contribute nothing and take the importer's mtime fallback) — prune so files deleted upstream, out of a local dir, or newly matching an ignore pattern 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) — and ``generate_ folder_summaries`` (phase 94, task 02) regenerates the stored folder summaries (the drill-down ``ls``'s per-level descriptions): its gate is the same change trigger — INCREMENTAL since phase 127 (task 03): the run's changed documents (``ImportSummary.changed_paths``, the import's added/updated/pruned ``(source, path)`` set) are passed as ``changed_docs=`` and ONLY their folder chains (each document's folder up to its source root — ``affected_folder_keys``) are rewritten, so a one-file change burns a handful of rows, not the whole table (the ``scripts/import_docs.py`` CLI keeps the byte-identical full regeneration — it passes no change set) **plus**, on an unchanged re-sync, a GAP probe — a candidate folder (at least 1 doc) with no stored row (``missing_folder_summaries``, phase 96, task 03): a gap fills ONLY the missing rows (``only_missing=True``; the old table-empty first-sync trigger is subsumed exactly — an empty table leaves every candidate missing), a complete table burns zero ``lite`` calls. It is per-folder fail-soft (a ``lite`` outage keeps the failed folders' previous rows — or leaves the row absent — and never flips the run to ``failed``) and only flushes — this run's own short-lived session commits (the phase-53 convention), so a folder failure never blocks step 6's bump; 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. Phase 134 (task 01) makes the machine FIVE-state: ``cancelled`` joins ``success`` / ``failed`` as a TERMINAL state — it carries ``finished_at``, clears ``current_file`` / ``phase`` / ``current_summary`` and the chunk position, and keeps the run's final ``files_done`` / ``files_total`` + summary counts (the phase-64/98 keep-final-counts convention) — but with ``error = None`` and ``detail = {}``: a cancel is intent, not failure, so the UI never treats it as an error (no banner, no modal — a neutral result line). ``GET /api/sync/status`` gains nothing structurally — the ``state`` key simply reports ``"cancelled"``. The cancel is COOPERATIVE, best-effort (phase 134 design): while a run is in flight, ``POST /api/sync/cancel`` (the router's ``require_admin`` dependency applies — 403 anonymous) sets the module-level ``_cancel_event`` AND ``_task.cancel()`` (unblocking whatever ``await`` the pipeline is parked on — e.g. a per-file LLM call inside the import), answering 202 the moment the request is REGISTERED (the run may already be terminal by the next poll — the status object, not the task, is authoritative). ``start_sync`` clears the event when it starts a run, so a stale event from a previously cancelled run can never bleed into the next one. ``_run_sync`` checks the event at the pipeline boundaries via :func:`_check_cancelled` (raises ``asyncio.CancelledError`` when set): the top of the per-source loop (before each clone/pull or local-dir check), immediately before ``import_sources``, and immediately before the overview + summaries span (the ``added + updated > 0`` block). Consequence a future reader must know: a git subprocess already running at the cancel moment runs to completion (the blocking ``subprocess.run`` has no await inside) — the cancel lands at the NEXT boundary (after that source's git ops, between files, or before overview/summaries); during the import span, the ``_task.cancel()`` from the endpoint interrupts at the next per-file await. The ``CancelledError`` contract has TWO paths (the former "deliberately not caught" rule refined, not broken): in ``_run_sync``'s ``except asyncio.CancelledError`` clause (placed before the ``except Exception``), a USER-requested cancel (``_cancel_event`` set) lands the run in the ``cancelled`` terminal + one INFO line (PLAN §9: ``sync: cancelled by user state=cancelled files_done=… files_total=…``) and then RE-RAISES (the task ends cancelled — ``_task.done()`` is True, the status object stays authoritative); an UNREQUESTED cancel (the app-shutdown path — the event unset) re-raises UNCHANGED (the shutdown contract holds: a background task must die in state, never as an unobserved exception — swallowing an unrequested stop would mask a real one). Phase 98 (task 01) adds the run's PHASE machine — ``phase`` is the pipeline stage the run is in: the model-check + clone/pull prelude reports ``null`` (the bare "Syncing…" label stays), then ``"import"`` (set immediately before ``import_sources``), ``"overview"`` (set before ``regenerate_overview`` — a changed KB only), and ``"summaries"`` (set before a folder-summary generation — the changed-KB incremental regeneration (phase 127, task 03 — the changed set's folder keys only) or the unchanged-walk gap-fill; the no-gap skip sets NO phase and the run stays ``"import"``). While in ``"summaries"`` the generator's ``on_progress`` hook fills ``current_summary`` (the ``source`` / ``source/folder_path`` being summarized — the bare source name for the source-root row) plus ``summaries_done`` / ``summaries_total`` — the per-folder position through the long summary span where the file count sits still (the hook's ``total`` is the run's key count — the AFFECTED count on a changed KB, the missing count on a gap-fill). Terminal states (success, failed, and — phase 134 — cancelled) clear ``phase`` + ``current_summary`` but keep the run's final ``summaries_done`` / ``summaries_total`` (the phase-64 keep-final-counts convention). Phase 129 (task 03) adds the in-flight file's EMBED-chunk position — ``current_chunk`` (the 1-based first chunk of the in-flight token-budgeted batch) + ``chunks_total`` (the file's total content chunks), batch granularity via the :meth:`app.rag.llm.LLMClient.embed` ``on_progress`` hook (tasks 01/02). They are 0/0 before the first batch (the prelude), during the ``"overview"`` / ``"summaries"`` phases, and in terminal states — the import phase clearing them at EVERY point it ends (the ``phase = "overview"`` and ``phase = "summaries"`` transitions and every terminal, the same spots ``current_file`` clears). Unlike the phase-64 FILE counts, the chunk position is a SUB-position of a file that's done — the keep-final- counts convention does NOT extend to it: the file counts survive the terminal for the result line's context, the chunk position clears with the file. 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.folder_summaries import generate_folder_summaries, missing_folder_summaries from app.rag.git_sources import clone_url_for, 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, file_commit_dates 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 five-state machine: ``idle`` (never run / reset), ``running``, ``success``, ``failed``, and ``cancelled`` (phase 134, task 01 — terminal like ``success`` / ``failed``: ``finished_at`` set, ``current_file`` / ``phase`` / ``current_summary`` cleared, the final file + summary counts kept, but ``error = None`` and ``detail = {}`` — a cancel is intent, not failure). 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). Phase 98 (task 01) phase fields: ``phase`` is the pipeline stage (null in the model-check + clone/pull prelude and in terminal states — the module docstring's phase machine); while in the ``"summaries"`` phase, ``current_summary`` is the source / ``source/folder_path`` the generator is summarizing right now and ``summaries_done`` / ``summaries_total`` carry the generator's progress hook's position — the hook's ``total`` is the run's key count: the AFFECTED count on a changed KB (phase 127, task 03 — the changed set's folder keys, NOT the table's candidate count) or the missing count on an unchanged-walk gap-fill (as before). Terminal states clear ``phase`` + ``current_summary`` but keep the run's final summary counts (the phase-64 keep-final-counts convention). Phase 129 (task 03) chunk-position fields: ``current_chunk`` is the 1-based first chunk of the in-flight token-budgeted batch and ``chunks_total`` the in-flight file's total content chunks (batch granularity, the ``LLMClient.embed`` hook — the module docstring's contract). They are 0/0 before the first batch, during the ``"overview"`` / ``"summaries"`` phases, and in terminal states — the import phase clears them at every point it ends (the overview/summaries transitions + every terminal, where ``current_file`` also clears). The phase-64 keep-final-counts convention covers the FILE counts only — the chunk position is a sub-position of a file that's done, so it clears in the terminal while ``files_done`` / ``files_total`` keep their final values. """ state: Literal["idle", "running", "success", "failed", "cancelled"] = "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 # Phase 98 (task 01): the phase machine — the pipeline stage the # run is in and, while in the folder-summary phase, the folder # being summarized plus the hook's done/total position. phase: Literal["import", "overview", "summaries"] | None = None current_summary: str | None = None summaries_done: int = 0 summaries_total: int = 0 # Phase 129 (task 03): the current file's EMBED chunk position — # the first chunk of the in-flight token-budgeted batch and the # file's total content chunks (the LLMClient.embed hook, batch # granularity). 0/0 before the first batch, during the # prelude/overview/summaries phases, and in terminal states. current_chunk: int = 0 chunks_total: int = 0 _status = SyncStatus() _task: asyncio.Task[None] | None = None # Phase 134 (task 01): the cancel request — set by POST # /api/sync/cancel while a run is in flight; _run_sync's boundaries # + its ``except CancelledError`` clause read it (the module # docstring's cancel paragraph). Cleared by start_sync when a run # starts, so a stale event can never bleed into the next run. _cancel_event = asyncio.Event() @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) and keep their final values in the terminal. ``current_chunk`` / ``chunks_total`` (phase 129) are the in-flight file's embed-chunk position (batch granularity — the first chunk of the in-flight batch over the file's total content chunks): 0/0 before the first batch, in the prelude/overview/ summaries phases, and in terminal states (they clear at every point the import phase ends; the FILE counts do not). ``phase`` (phase 98) is the pipeline stage — null in the prelude and terminal states; while in ``"summaries"``, ``current_summary`` + ``summaries_done`` / ``summaries_total`` carry the generator's position (the terminal keeps the final summary counts). Phase 134 (task 01): ``state`` also reports ``"cancelled"`` for a user-cancelled run — a terminal state with ``error`` null and ``detail`` empty; the run's final file + summary counts are kept, the chunk position cleared. """ 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, # Phase 129 (task 03): the in-flight file's embed-chunk # position — the dataclass defaults make idle 0/0, and the # runner clears it at every point the import phase ends # (overview/summaries transitions + every terminal). "current_chunk": _status.current_chunk, "chunks_total": _status.chunks_total, # Phase 98 (task 01): the phase machine — null/0/0 idle (the # dataclass defaults) and in terminal states (which clear # phase + current_summary but keep the final summary counts). "phase": _status.phase, "current_summary": _status.current_summary, "summaries_done": _status.summaries_done, "summaries_total": _status.summaries_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). Phase 134 (task 01): the run's cancel request resets with the run — ``_cancel_event.clear()`` before the task is created, so a stale event from a previously cancelled run can never bleed into this one. """ global _task if _task is not None and not _task.done(): raise HTTPException(status_code=409, detail="a sync is already running") _cancel_event.clear() _task = asyncio.create_task(_run_sync()) return {"detail": "sync started"} @router.post("/cancel", status_code=202) async def cancel_sync() -> dict[str, str]: """Request the cancel of the in-flight run (phase 134, task 01). 202 + ``cancel requested`` is answered the moment the request is REGISTERED: the module ``_cancel_event`` is set (the runner's ``_check_cancelled`` boundaries + its ``except CancelledError`` conversion see it) AND ``_task.cancel()`` unblocks whatever ``await`` the pipeline is parked on (e.g. a per-file LLM call inside the import). The run may already be terminal by the next status poll — the status object, not the task, is authoritative, so the UI's poll settles on the terminal (``cancelled``) state. 409 ``no sync is running`` when there is no run in flight (the ``_task`` is None or done — the harmless race the UI's poll settles; no error copy, no state change). """ if _task is None or _task.done(): raise HTTPException(status_code=409, detail="no sync is running") _cancel_event.set() _task.cancel() return {"detail": "cancel requested"} def _check_cancelled() -> None: """Phase 134 (task 01): the cooperative cancel boundary. Raises ``asyncio.CancelledError`` when the module ``_cancel_event`` is set (the user requested a stop — ``POST /api/sync/cancel``); :func:`_run_sync`'s ``except asyncio.CancelledError`` clause converts it into the ``cancelled`` terminal state. Called at the pipeline boundaries (the module docstring's cancel paragraph) — the top of the per-source loop, before ``import_sources``, and before the overview + summaries span. A git subprocess in flight at the cancel moment runs to completion (no await inside ``subprocess.run``); the cancel lands at the NEXT boundary. """ if _cancel_event.is_set(): raise asyncio.CancelledError() 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. Cancellation (phase 134, task 01 — the module docstring's cancel paragraph): the cooperative boundaries (:func:`_check_cancelled` at the top of the per-source loop, before ``import_sources``, and before the overview + summaries span) plus the ``except asyncio.CancelledError`` clause convert a USER-requested cancel (``_cancel_event`` set by ``POST /api/sync/cancel``) into the ``cancelled`` terminal state + one INFO line, then re-raise (the task ends cancelled — the status object stays authoritative). An UNREQUESTED cancel (the app-shutdown path — the event unset) re-raises unchanged: a background task is never swallowed, and swallowing a real stop would mask it. """ _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 # Phase 98 (task 01): the phase fields reset with the run — no # phase until the import starts (the model-check + clone/pull # prelude reports null), no folder until the summary span starts. _status.phase = None _status.current_summary = None _status.summaries_done = 0 _status.summaries_total = 0 # Phase 129 (task 03): the chunk-position fields reset with the # run — 0/0 until the import's first embed batch reports. _status.current_chunk = 0 _status.chunks_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] = [] ignore_by_root: dict[str, list[str]] = {} include_hidden_by_root: dict[str, bool] = {} doc_dates_by_root: dict[str, dict[str, datetime]] = {} for row in rows: # Phase 134 (task 01): the per-source boundary — a cancel # registered while the previous source's git ops run (a # blocking subprocess runs to completion) lands here, # before this source's clone/pull or local-dir check. _check_cancelled() if row.kind == "git": # Phase 121: the token column is injected into the clone # URL ONLY here (clone_url_for — NULL token → the bare # stored URL verbatim); repo_name stays on the bare URL # so the checkout directory name is credential-free. # Phase 135: the row's branch pin is the third # argument — NULL (every pre-phase-135 row + every # env-fallback row) keeps the pre-phase argv # byte-identical; a pinned branch is checked out on a # fresh clone (--branch) and guarded + tracked on an # existing checkout (fetch + ff-merge of that ref). root = clone_or_pull( clone_url_for(row), sources_root / repo_name(row.url), row.branch, ) # Phase 106 (D2): the checkout's per-file last-commit # dates, keyed by the SAME root string the importer # sees (full-history checkouts → true per-file # last-commit dates for every git source — phase 107 # D11 supersedes phase 106 D10's shallow tip-date # behavior). Local rows contribute nothing — the # importer's mtime fallback applies to them. doc_dates_by_root[str(root)] = file_commit_dates(root) 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. root = Path(row.path or row.url).expanduser() if not root.is_dir(): raise GitSyncError(f"local source missing: {root}") sources.append(root) # Phase 89: the row's ignore list, keyed by the SAME root # string the importer sees; two rows sharing a root string # get the union (extend, not replace) — the sibling/repo-name # edge. if row.ignore_paths: ignore_by_root.setdefault(str(root), []).extend(row.ignore_paths) # Phase 105 (A1/A4): the row's hidden-folders flag, keyed by # the SAME root string the importer sees; a shared-root # collision ORs — if EITHER row says "index hidden", the # root does (the ignore-map union's boolean mirror). include_hidden_by_root[str(root)] = ( include_hidden_by_root.get(str(root), False) or bool(row.include_hidden) ) # 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 # Phase 129 (task 03): the in-flight file's embed-chunk # position — the LLMClient.embed hook (task 01) fires before # each token-budgeted batch with (first_chunk_1based, total); # the file's IDENTITY is the sibling ``progress`` hook's (the # two hooks are paired — a progress call names the file, and # every chunk position until the next one belongs to it). def _chunk_hook(chunk: int, total: int) -> None: _status.current_chunk = chunk _status.chunks_total = total # Phase 98 (task 01): the phase machine — the import phase # starts NOW (the model-check + clone/pull prelude above # reported ``phase: null``). _status.phase = "import" # Phase 134 (task 01): the import boundary — a cancel between # the source walk and the import (or parked in a blocking git # span that just finished) lands here, before any file is # imported. _check_cancelled() summary: ImportSummary = await import_sources( sources, llm, prune=True, progress=_hook, on_chunk_progress=_chunk_hook, ignore_by_root=ignore_by_root, include_hidden_by_root=include_hidden_by_root, doc_dates_by_root=doc_dates_by_root, ) # Phase 134 (task 01): the overview + summaries boundary — a # cancel during the import span (the endpoint's _task.cancel() # interrupts at the next per-file await) or registered anywhere # before this point lands here, before the lite spans start. _check_cancelled() overview = False if summary.added + summary.updated > 0: _status.phase = "overview" # phase 98 (task 01) # Phase 129 (task 03): the import phase ends here — the # chunk position (a sub-position of a done file) clears # with the phase change; the file counts keep their final # values (the phase-64 convention covers files only). _status.current_chunk = 0 _status.chunks_total = 0 overview = await regenerate_overview(llm) # Phase 94 (task 02), phase 96 (task 03), phase 127 (task 03): # the folder summaries — the drill-down ls's per-level # descriptions. A changed KB (added + updated > 0) is an # INCREMENTAL regeneration (phase 127, task 03): the run's # changed documents (summary.changed_paths — the import's # added/updated/pruned set) are passed as changed_docs= and # ONLY their folder chains (each document's folder up to its # source root) are rewritten — a one-file change burns a # handful of rows, not the whole table (the CLI keeps the # byte-identical full regeneration — it passes no change set); # an unchanged re-sync takes the GAP probe instead of the old # table-empty one — a candidate folder (at least 1 doc — every # existing folder) with no stored row: a gap # fills ONLY the missing rows (only_missing=True — the # subsumed table-empty first-sync trigger included, where # every candidate is missing), a complete table burns zero # lite calls. The generator is per-folder fail-soft (a lite # outage never flips the run to failed) and only flushes: this # run's own short-lived session commits (the phase-53 # convention), so the step-6 bump stays change-gated on the KB, # not on the summaries. The status surface is the PHASE # machine (phase 98, task 01): the stats dict itself stays # log-only (the detail shape is untouched). # Phase 98 (task 01): the folder-summary progress hook — the # closure-captures-``_status`` convention the file's ``_hook`` # above already uses; the generator's ``total`` (the loop-start # key count — the affected count on a changed KB, the missing # count on a gap-fill) and the bare source name for the root # row come straight from the hook's arguments (D5's shape). def _summary_hook(done: int, total: int, source: str, folder_path: str) -> None: _status.current_summary = ( source if folder_path == "" else f"{source}/{folder_path}" ) _status.summaries_done = done _status.summaries_total = total fs_db = SessionLocal() try: if summary.added + summary.updated > 0: # Changed KB: incremental regeneration (phase 127, # task 03) — ONLY the changed set's folder keys # (changed_docs=summary.changed_paths: each document's # folder chain up to its source root); the summaries # phase starts now, and the hook reports the per-folder # position through the (long) span (its total is the # affected count). _status.phase = "summaries" # Phase 129 (task 03): the import phase ends here too # (the changed-KB branch skips no phase) — same # chunk-position clear as the overview transition. _status.current_chunk = 0 _status.chunks_total = 0 folder_stats = await generate_folder_summaries( fs_db, llm, changed_docs=summary.changed_paths, on_progress=_summary_hook, ) fs_db.commit() logger.info( "sync: folder_summaries changed=%d stats=%s", len(summary.changed_paths), folder_stats, ) else: missing = missing_folder_summaries(fs_db) if missing: # Unchanged-walk gap-fill: the SAME summaries # phase + hook (the fill's total is the missing # count — the hook's loop-start ``total``). _status.phase = "summaries" # Phase 129 (task 03): the import phase ends here # too (the unchanged-walk gap-fill branch) — same # chunk-position clear. _status.current_chunk = 0 _status.chunks_total = 0 folder_stats = await generate_folder_summaries( fs_db, llm, only_missing=True, on_progress=_summary_hook ) fs_db.commit() logger.info( "sync: folder_summaries gap-fill missing=%d stats=%s", len(missing), folder_stats, ) else: # The no-gap skip sets NO phase — the run stays # "import" through to the terminal (the phase-64 # bare-label pin holds for this shape). logger.info("sync: folder_summaries skipped (KB unchanged)") finally: fs_db.close() # 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.phase = None # phase 98: keep the final summary counts _status.current_summary = None # Phase 129 (task 03): the chunk position is a sub-position of # a file that's done — it clears in the terminal (unlike the # FILE counts, which keep their final values for the result # line's context). _status.current_chunk = 0 _status.chunks_total = 0 _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, "dates_updated": summary.dates_updated, "overview": overview, "sources_version": sources_version, } logger.info("sync: done detail=%s", _status.detail) except asyncio.CancelledError: # Phase 134 (task 01): the TWO conversion paths (the module # docstring's cancel paragraph). A user-requested cancel — # ``_cancel_event`` set by POST /api/sync/cancel, whether the # endpoint's ``_task.cancel()`` interrupted a parked await or a # ``_check_cancelled`` boundary raised — lands the run in the # ``cancelled`` terminal and re-raises (the task ends # cancelled — ``_task.done()`` is True, the status object stays # authoritative). An unrequested cancel (the app-shutdown path # — the event unset) re-raises unchanged: never swallow a real # stop. if _cancel_event.is_set(): _status.state = "cancelled" _status.finished_at = datetime.now(UTC) _status.error = None # a cancel is intent, not failure _status.current_file = None # keep the final counts _status.phase = None # keep the final summary counts _status.current_summary = None # Phase 129 (task 03): the chunk position is a sub-position # of a file that's done — it clears in EVERY terminal, like # success and failed (the convention split, pinned). _status.current_chunk = 0 _status.chunks_total = 0 logger.info( "sync: cancelled by user state=cancelled files_done=%d files_total=%d", _status.files_done, _status.files_total, ) raise 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 _status.phase = None # phase 98: keep the final summary counts _status.current_summary = None # Phase 129 (task 03): the chunk position clears in the # failed terminal too (the convention split — the FILE counts # keep their final values next to the error, the sub-position # of a file that's done/abandoned does not). _status.current_chunk = 0 _status.chunks_total = 0