"""Knowledge-base importer (PLAN §5 / §9 / §11).

Walks the in-scope files (the A9 family by default — the original seven
plus the quadlet family and ``j2`` — ``BOR_IMPORT_EXTENSIONS``, which may
name any well-formed extension; case-insensitive), diffs by sha256
against ``documents.content_hash`` and, for every new or changed file, runs
the two-phase upsert:

1. upsert the document row and replace its chunk rows (embeddings NULL)
2. embed the new chunks in batches and attach the vectors
3. commit — one transaction per file, so a failed embedding leaves the
   database untouched and the file is simply retried on the next run
4. every file (phase 30; phase 118, A2: markdown included — the
   non-markdown-only scope is retired): generate a ``lite``-model summary
   and, best-effort, store it on ``documents.summary`` plus one extra
   embedded chunk (``is_summary``, position −1). The document row and its
   content chunks are already committed at this point, so a summary
   failure only means the file is indexed without a summary (logged and
   counted in ``summary_errors``) — it is never lost.

Scope (A9, revised 2026-08-21): any path containing a dot-prefixed
component (hidden dirs — vendored caches like ``.esphome/.espressif/**`` —
or hidden files) is skipped, plus the well-known exclusion list — UNLESS
the source's phase-105 hidden-folders flag admits dot-prefixed paths; the
exclusion list always applies. A token may also name extensionless files
by their exact lowercased full filename (``Dockerfile`` under the
``dockerfile`` token — phase 102).

``prune=True`` deletes documents (of the imported sources only) whose files
no longer exist **or no longer match the format filter** — this is how
previously-imported junk (e.g. dot-dir READMEs) leaves the index. Per-file
logging uses the verbs ``added | updated | unchanged | pruned`` plus a
summary line with per-format counts (PLAN §9). Phase 122 prune guard
(LOCKED, derived from A3/A4): while the ``images`` toggle is OFF, an
``is_image`` doc is INVISIBLE to the walk, not a deleted file — prune
skips it (turning the toggle off and syncing must never destroy image
documents); a toggle-ON run prunes a deleted image file normally and
deletes its ``image_dir`` copy with the row.

Document dates (phase 106, D2/D4): every import sources
``documents.created_at`` from the file's source — the per-file git
last-commit date when a ``doc_dates_by_root`` entry names the file,
else the file's mtime — normalized by
:func:`app.rag.doc_dates.normalize_doc_date` (undetermined or future →
today, D3) on every add and update. On the unchanged path the stored
date is REFRESHED from the same source (it may go OLDER — no monotonic
guard) and counted in ``summary.dates_updated`` — unless the row
carries the owner's manual correction (``created_at_manual``, D1), which
the sync never touches.

NULL-summary backfill (phase 118, A2): an UNCHANGED file (same
``content_hash``) whose ``documents.summary`` is still NULL — a
pre-phase-30 row, or an earlier fail-soft miss — gets the same
best-effort summary pass on every sync until it sticks. A success counts
``summary_backfilled`` (never ``summaries``) and touches nothing else: no
content re-embed, no added/updated/pruned count — so no
``sources_meta`` bump, no KB-overview/folder-summary regeneration. The
backfill runs BEFORE the ``created_at_manual`` early-return (the manual
flag protects the DATE only, D1) and the strict ``is None`` check leaves
owner-set summaries (even empty strings, phase 57) alone.

Standalone images (phase 122, LOCKED A3): with the ``images`` toggle
(``BOR_IMAGES``) ON, the walk also admits the image extension set
(``BOR_IMAGE_EXTENSIONS`` — a SEPARATE set from ``import_extension_set``;
images are never user-added via ``BOR_IMPORT_EXTENSIONS``, the toggle is
the single knob). Such a file takes the binary index path
(:func:`_index_image_file`): the sha256 digest is over the raw BYTES
(content identity — the digest rule is unchanged), the bytes are copied
to the persistent home ``settings.image_dir/<doc-id>.<ext>`` (dir created
on demand; the copy is written ONLY after a successful description, so a
failure never leaves an orphan; a changed image deletes the stale copy
first; a pruned image doc deletes its copy), the row carries
``is_image=True`` + ``image_path``, and ``content`` is the vision
description — the ONLY embedded text of the document (the embedding model
never sees pixels; ``read_text`` is never called for an image). The
description comes through the single seam :func:`_describe_or_skip`
(task 03: :func:`app.rag.summarizer.describe_image` — ONE CHAT-model
(vision) call with the image bytes as a base64 data URL; the ``lite``
summary model is NOT assumed vision-capable, LOCKED A3); a failed/empty
description SKIPS the doc entirely (no row, no copy) — counted in
``images_failed`` + a warning, the sync continues (fail-soft). The normal
chunk pipeline then embeds ``content`` and the phase-30 summary path runs
on it — image-aware (task 03): for an image doc the description IS the
summary (stored verbatim, no ``lite`` call, no pointer line), so the
``is_summary`` position −1 chunk mirrors ``Document.summary``, which
equals ``Document.content``.

``import_sources`` accepts an optional per-file ``progress`` callback
(phase 64, task 01) reporting the file being processed right now.
"""
from __future__ import annotations

import hashlib
import logging
import uuid
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Protocol

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.config import Settings
from app.db import SessionLocal
from app.models import Chunk, Document
from app.rag.chunker import chunk_document, extract_title
from app.rag.doc_dates import file_mtime_datetime, normalize_doc_date
from app.rag.llm import EmbeddingError, LLMError
from app.rag.summarizer import (
    IMAGE_FALLBACK_MIME,
    IMAGE_MIMES,
    describe_image,
    generate_summary,
)

logger = logging.getLogger("app.importer")

#: Non-content directories never imported (PLAN anchor A9).
EXCLUDED_DIRS: frozenset[str] = frozenset(
    {".venv", "node_modules", ".git", "__pycache__", ".pytest_cache", "dist", "build"}
)


class Embedder(Protocol):
    """Everything the importer needs from the LLM client (duck-typed for tests)."""

    settings: Settings
    embed_batches: int

    async def embed(self, texts: list[str]) -> list[list[float]]: ...

    async def chat(self, messages: list[dict[str, Any]], model: str | None = None) -> str: ...
    # ^ the one-shot completion the summarizer uses — the ``lite`` model
    #   for text summaries (phase 30, task 01) and the CHAT (vision)
    #   model for the phase-122 image description (multimodal content:
    #   a string or a list of OpenAI-compatible parts); :class:`app.rag.
    #   llm.LLMClient` satisfies it.


@dataclass
class ImportSummary:
    """Counts for one import run (also printed by the CLI)."""

    files: int = 0
    added: int = 0
    updated: int = 0
    unchanged: int = 0
    pruned: int = 0
    errors: int = 0
    chunks: int = 0
    embed_batches: int = 0
    #: Files whose lite summary was generated + indexed (phase 30; phase
    #: 118, A2: every A9 format, markdown included). One ``is_summary``
    #: chunk per success.
    summaries: int = 0
    #: Files whose summary generation failed (best-effort — the document
    #: is still indexed, without a summary).
    summary_errors: int = 0
    #: Unchanged docs whose NULL summary was backfilled (phase 118, A2) —
    #: one ``is_summary`` chunk per success; the content is untouched, so
    #: a backfill NEVER counts added/updated/pruned (no
    #: ``sources_meta`` bump, no overview/folder-summary regeneration).
    summary_backfilled: int = 0
    #: Files whose ``created_at`` was refreshed on the UNCHANGED path —
    #: content untouched, date re-sourced (phase 106, D4: the date may
    #: go OLDER; a date-only refresh NEVER counts added/updated/pruned,
    #: so no ``sources_meta`` bump, no overview/folder-summary
    #: regeneration).
    dates_updated: int = 0
    #: Image docs (phase 122, LOCKED A3) whose vision description failed
    #: or came back empty — the doc is SKIPPED entirely (no row, no
    #: ``image_dir`` copy): an undescribed image is unsearchable noise.
    #: Fail-soft: the sync continues, this counter + the warning line
    #: are the signal.
    images_failed: int = 0
    #: Files walked, keyed by lowercased extension (``md``, ``yaml``, …).
    formats: dict[str, int] = field(default_factory=dict)

    def format_counts(self) -> str:
        """``md:203,yaml:267,py:14`` — highest count first (PLAN §9)."""
        if not self.formats:
            return "none"
        ordered = sorted(self.formats.items(), key=lambda kv: (-kv[1], kv[0]))
        return ",".join(f"{ext}:{count}" for ext, count in ordered)

    def log(self) -> None:
        logger.info(
            "import: summary files=%d added=%d updated=%d unchanged=%d pruned=%d "
            "errors=%d chunks=%d embed_batches=%d summaries=%d summary_errors=%d "
            "summary_backfilled=%d dates_updated=%d images_failed=%d formats=%s",
            self.files,
            self.added,
            self.updated,
            self.unchanged,
            self.pruned,
            self.errors,
            self.chunks,
            self.embed_batches,
            self.summaries,
            self.summary_errors,
            self.summary_backfilled,
            self.dates_updated,
            self.images_failed,
            self.format_counts(),
        )


def normalize_ignore_path(entry: str) -> str:
    """One ignore-path entry → canonical form (phase 89, A1).

    Trim surrounding whitespace, then strip ALL leading/trailing
    ``/`` — so ``"/my/files/"``, ``"my/files/"`` and ``"my/files"``
    all become ``"my/files"``. ``""`` / ``"//"``,
    ``"  "`` normalize to ``""`` (callers drop empties).
    """
    return entry.strip().strip("/")


def is_ignored(rel: str, prefixes: tuple[str, ...]) -> bool:
    """Phase 89, A1 — the pure prefix rule, nothing else.

    ``rel`` is the source-relative POSIX path WITHOUT a leading
    slash (the ``documents.path`` string). Match = ``rel`` STARTS
    WITH a normalized entry: raw string prefix — deliberately NO
    component-boundary check (``"my/files"`` also matches
    ``"my/files2/x.md"``) and NO mid-path matching (``"myfile.txt"``
    matches ``"myfile.txt"`` but not ``"some/path/myfile.txt"``).
    """
    return any(rel.startswith(p) for p in prefixes)


def _ignore_for_root(
    root: Path, ignore_by_root: dict[str, list[str]] | None
) -> tuple[str, ...]:
    """The normalized, non-empty prefix tuple for one root (phase 89).

    Keyed by ``str(root)`` — the root string exactly as the caller
    passed it in ``sources`` (unambiguous when two rows share a
    source *name* but different dirs). Callers may pass RAW box
    lines: the importer normalizes + drops empties here, the single
    choke point — stored lists (already normalized) normalize to
    themselves.
    """
    raw = (ignore_by_root or {}).get(str(root)) or []
    return tuple(p for p in (normalize_ignore_path(e) for e in raw) if p)


def _include_hidden_for_root(
    root: Path, include_hidden_by_root: dict[str, bool] | None
) -> bool:
    """The per-root hidden-folders flag (phase 105, A1).

    Keyed by ``str(root)`` — the root string exactly as the caller
    passed it in ``sources`` (the ``_ignore_for_root`` convention,
    phase 89): ``True`` only for roots the caller lists as True;
    unlisted/``None`` roots are ``False`` — every existing caller
    behaves byte-identically (A4).
    """
    return bool((include_hidden_by_root or {}).get(str(root), False))


def match_extension(path: Path, extensions: frozenset[str]) -> str | None:
    """The bare lowercased token *path* imports under, or ``None``.

    1. Non-empty lowercased dotted suffix in *extensions* (the A9 rule —
       ``kubernetes.md`` → ``md``).
    2. No suffix: the lowercased FULL filename equals a bare token of
       *extensions* (``Dockerfile`` → ``dockerfile``) — the phase-102
       extensionless rule. Exact name only: ``mydockerfile`` never
       matches the ``dockerfile`` token.
    """
    suffix = path.suffix.lower()
    if suffix and suffix in extensions:
        return suffix.lstrip(".")
    if not suffix and path.name.lower() in {e.lstrip(".") for e in extensions}:
        return path.name.lower()
    return None


def iter_importable_files(
    root: Path,
    extensions: frozenset[str],
    excluded: frozenset[str] = EXCLUDED_DIRS,
    ignore: tuple[str, ...] = (),
    include_hidden: bool = False,
    image_extensions: frozenset[str] = frozenset(),
) -> list[Path]:
    """All importable files under *root* (sorted), per the A9 scope rules.

    *extensions* is a set of lowercased dotted suffixes (``{'.md', '.py'}``).
    Skips: when ``include_hidden`` is False (the default), any path with a
    dot-prefixed component (hidden dirs/files — vendored caches like
    ``.esphome/.espressif/**``); when True, dot-prefixed components are
    ADMITTED (files inside hidden folders, and hidden files) and only
    *excluded* is consulted (A1 — caches/VCS internals are never content).
    The well-known non-content directories in *excluded* are skipped in
    BOTH states. *ignore* (phase 89, A1) is a
    tuple of ALREADY-normalized, non-empty source-relative path prefixes
    (the importer's ``_ignore_for_root`` is the normalization choke point
    — raw box lines never reach this function): a file is skipped when its
    source-relative POSIX path starts with any entry; the default ``()``
    keeps every existing caller byte-identical. The *ignore* tuple
    composes additively in both states.

    *image_extensions* (phase 122) is the lowercased dotted
    image-extension set admitted IN ADDITION to *extensions* — passed by
    :func:`import_sources` only while the ``images`` toggle is on (it
    reads ``llm.settings``; the image set is never merged into
    *extensions*). The empty default admits nothing: every existing caller
    (and the toggle-off walk) stays byte-identical to pre-phase.
    """
    if not root.is_dir():
        return []
    files: list[Path] = []
    for path in sorted(root.rglob("*")):
        if not path.is_file():
            continue
        rel = path.relative_to(root)
        if any(
            (not include_hidden and part.startswith(".")) or part in excluded
            for part in rel.parts
        ):
            continue
        if ignore and is_ignored(rel.as_posix(), ignore):
            continue
        matched = match_extension(path, extensions)
        if matched is None and image_extensions:
            # Phase 122: the image set is admitted IN ADDITION to the
            # import set (toggle on only — the caller passes it in).
            matched = match_extension(path, image_extensions)
        if matched is None:
            continue
        files.append(path)
    return files


async def import_sources(
    sources: list[Path],
    llm: Embedder,
    *,
    prune: bool = False,
    limit: int | None = None,
    session: Session | None = None,
    progress: Callable[[str, str, int, int], None] | None = None,
    ignore_by_root: dict[str, list[str]] | None = None,
    include_hidden_by_root: dict[str, bool] | None = None,
    doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,
) -> ImportSummary:
    """Import every A9-format file under *sources* (see module docstring).

    ``session`` may be supplied (tests); a private one is opened and closed
    otherwise. ``limit`` caps the number of files processed (debug only) and
    disables pruning, since an incomplete walk must not drive deletions.

    ``progress`` (phase 64, task 01) is an optional per-file hook called
    once per importable file, immediately before that file's
    ``_index_file`` — with ``(source, rel_posix_path, done, total)``: the
    same POSIX *rel* the document rows use, ``done`` = the 1-based index of
    the current file **across all sources**, and ``total`` = the combined
    pre-walk count of importable files across all *sources* roots. The
    pre-walk (same extension/exclusion rules, directory stats only, no file
    reads) happens **only when *progress* is provided**: callers passing
    nothing pay no extra walk and behave exactly as before. Under
    ``limit``, the hook still fires per processed file only — ``done`` never
    exceeds the limit, but ``total`` stays the full pre-walk count (an
    incomplete walk must not misreport the denominator).

    ``ignore_by_root`` (phase 89, A1/A2) maps ``str(root)`` — the root path
    string exactly as passed in *sources* — to that source's RAW ignore-path
    lines (the importer normalizes them via ``_ignore_for_root``, the single
    choke point): matching files are never walked, so they are never
    embedded and never summarized, and the progress pre-walk uses the same
    per-root tuple as the processing loop, so ``total`` never counts them.
    A file that newly matches a pattern simply never enters ``seen``, so the
    next ``prune=True`` run deletes its row automatically (A2 — the A9
    junk-precedent). ``None`` (the default) changes nothing: the map is read
    per root, unlisted roots get an empty tuple, and every existing caller
    behaves byte-identically.

    ``include_hidden_by_root`` (phase 105, A1) maps ``str(root)`` to the
    stored flag: ``True`` admits dot-prefixed components for that root
    (``EXCLUDED_DIRS`` and the extension filter still apply; the ignore
    tuple composes additively). Unlisted/``None`` roots are ``False`` —
    byte-identical to pre-phase-105. A file that was indexed with the
    flag ON and is walked again with it OFF simply never enters
    ``seen``, so the next ``prune=True`` run deletes its row
    automatically (A2 — the A9/phase-89 precedent).

    ``doc_dates_by_root`` (phase 106, D2/D4) maps ``str(root)`` — the
    root path string exactly as passed in *sources* — to that source's
    RAW per-file source dates: source-relative POSIX path → the git
    last-commit datetime (task 03's ``file_commit_dates``). ONLY git
    roots are listed — unlisted roots (local dirs, unpacked uploads)
    take the mtime fallback, and a path missing from its root's map
    does too. The map entry beats the file's mtime when present. The
    progress pre-walk is untouched (dates change no file count).
    ``None`` (the default) changes nothing for existing callers: the
    mtime fallback applies to every file — which IS the behavior
    change, D4: an unchanged file now refreshes its stored date from
    its source on every run (the backfill-correction case).

    Images (phase 122): when ``llm.settings.images`` is on, BOTH walks
    (the progress pre-walk and the processing loop — same rules, so
    ``total`` counts images) also admit ``llm.settings.image_extension_set``
    files, each indexed through the binary image path (see the module
    docstring). ``prune=True`` with the toggle ON prunes a deleted image
    file normally (row + ``image_dir`` copy); with the toggle OFF the
    prune skips ``is_image`` docs (the prune guard — the image is
    invisible to the walk, not a deleted file).
    """
    if limit is not None and limit <= 0:
        raise ValueError("limit must be >= 1")
    summary = ImportSummary()
    # Phase 122: the image set is admitted by the walks ONLY while the
    # toggle is on — the empty set admits nothing, so the toggle-off run
    # (walk, counts, prune) stays byte-identical to pre-phase.
    image_exts = llm.settings.image_extension_set if llm.settings.images else frozenset()
    owns_session = session is None
    if session is None:
        session = SessionLocal()
    seen: set[tuple[str, str]] = set()
    source_names: set[str] = set()
    # phase 64 (task 01): the hook's combined denominator, walked with the
    # exact same rules as the processing loop below (directory stats only,
    # no file reads). Skipped entirely for ``progress=None`` callers — no
    # extra pass, byte-identical behaviour and cost.
    total = 0
    if progress is not None:
        for root in sources:
            total += len(
                iter_importable_files(
                    root,
                    llm.settings.import_extension_set,
                    ignore=_ignore_for_root(root, ignore_by_root),
                    include_hidden=_include_hidden_for_root(
                        root, include_hidden_by_root
                    ),
                    image_extensions=image_exts,
                )
            )
    try:
        for root in sources:
            if not root.is_dir():
                logger.warning("import: source dir not found, skipping: %s", root)
                continue
            if limit is not None and summary.files >= limit:
                break
            source = root.name
            source_names.add(source)
            ignore = _ignore_for_root(root, ignore_by_root)
            include_hidden = _include_hidden_for_root(root, include_hidden_by_root)
            # Phase 106 (D2): the root's raw source dates (git last-commit
            # for git roots, keyed by the same str(root) convention); {}
            # for unlisted roots — every file then takes the mtime fallback.
            dates_map = (doc_dates_by_root or {}).get(str(root), {})
            for path in iter_importable_files(
                root,
                llm.settings.import_extension_set,
                ignore=ignore,
                include_hidden=include_hidden,
                image_extensions=image_exts,
            ):
                if limit is not None and summary.files >= limit:
                    break
                rel = path.relative_to(root).as_posix()
                seen.add((source, rel))
                summary.files += 1
                # Phase 102: the matched bare token (``dockerfile`` for an
                # extensionless ``Dockerfile``), never ``unknown`` — the
                # file is in scope, so the walk matched it. Phase 122: an
                # image file matches the image set, not the import set.
                ext = (
                    match_extension(path, llm.settings.import_extension_set)
                    or match_extension(path, image_exts)
                    or "unknown"
                )
                summary.formats[ext] = summary.formats.get(ext, 0) + 1
                if progress is not None:
                    # phase 64: report the file *before* indexing it — a
                    # file that then errors or turns out unchanged was
                    # already the "current file". No try/except around the
                    # call: the hooks in this repo only assign fields.
                    progress(source, rel, summary.files, total)
                try:
                    await _index_file(
                        session, source=source, rel=rel, full_path=path, llm=llm,
                        summary=summary, raw_date=dates_map.get(rel),
                    )
                except EmbeddingError as e:
                    # A pathological file (e.g. content the embedding endpoint
                    # refuses) must not abort the whole KB: roll back its
                    # uncommitted rows, log loudly, and keep going. The next
                    # run retries it.
                    session.rollback()
                    summary.errors += 1
                    logger.error("import: error source=%s path=%s — %s", source, rel, e)
        if prune:
            if limit is not None:
                logger.warning("import: --prune ignored because --limit was given")
            else:
                summary.pruned = _prune(
                    session, source_names, seen, images=llm.settings.images
                )
        summary.embed_batches = llm.embed_batches
        summary.log()
        return summary
    finally:
        if owns_session:
            session.close()


async def _index_file(
    session: Session,
    *,
    source: str,
    rel: str,
    full_path: Path,
    llm: Embedder,
    summary: ImportSummary,
    raw_date: datetime | None = None,
) -> None:
    """Upsert one file: doc row + chunk rows + embeddings, one transaction.

    ``raw_date`` (phase 106, D2) is the file's RAW source date — the
    git last-commit datetime from the caller's ``doc_dates_by_root``
    map, or ``None`` (every non-git case): the file's mtime is read
    here, once, and becomes the source date (the D2 fallback).

    Image files (phase 122, toggle on) delegate to
    :func:`_index_image_file` — the binary path (bytes digest,
    persistent copy, ``content`` = the vision description) — BEFORE
    any text read: ``read_text`` is never called for an image.
    """
    settings = llm.settings
    # Phase 122 (task 02): the image branch FIRST. Only reachable while
    # the ``images`` toggle is on — the walk never admits image files
    # while it is off, and with it off this check is a no-op (the text
    # path below stays byte-identical to pre-phase).
    if settings.images:
        image_set = settings.image_extension_set
        if match_extension(full_path, image_set) is not None:
            return await _index_image_file(
                session, source=source, rel=rel, full_path=full_path, llm=llm,
                summary=summary, raw_date=raw_date,
            )
    content = full_path.read_text(encoding="utf-8", errors="replace").replace("\x00", "")
    digest = hashlib.sha256(content.encode("utf-8")).hexdigest()
    doc = session.scalar(select(Document).where(Document.source == source, Document.path == rel))
    if raw_date is None:
        # D2 fallback: no source date in the map → the file's mtime
        # (one stat). Read before the unchanged early-return — the
        # unchanged path refreshes the stored date from the same source.
        raw_date = file_mtime_datetime(full_path)
    if doc is not None and doc.content_hash == digest:
        summary.unchanged += 1
        logger.info("import: unchanged source=%s path=%s", source, rel)
        # Phase 118 (A2): an unchanged doc whose summary is still NULL
        # (a pre-phase-30 row, or an earlier fail-soft miss) gets a
        # summary-only backfill — one ``is_summary`` chunk, no content
        # re-embed, and NEVER an added/updated/pruned count (so no
        # ``sources_meta`` bump, no overview/folder-summary
        # regeneration). Strict ``is None``: an empty-string summary is
        # owner-set (phase 57) and is never overwritten. BEFORE the
        # manual-date early-return: ``created_at_manual`` protects the
        # DATE only (phase 106, D1), not the summary.
        if doc.summary is None:
            await _store_summary(
                session, doc=doc, source=source, rel=rel, content=content,
                llm=llm, summary=summary, backfill=True,
            )
        if doc.created_at_manual:
            # D1/D4: the owner's correction survives the sync — no
            # write at all (the phase-97 ``manually_edited`` precedent).
            return
        # D4: the date refreshes on every sync, including unchanged
        # files, and may go OLDER (no monotonic guard). A date-only
        # refresh is still counted ``unchanged`` — never added/updated/
        # pruned, so no ``sources_meta`` bump and no regeneration.
        target = normalize_doc_date(raw_date)
        if target != doc.created_at:
            doc.created_at = target
            session.commit()
            summary.dates_updated += 1
            logger.info(
                "import: date-refreshed source=%s path=%s date=%s",
                source, rel,
                doc.created_at.isoformat(),
            )
        return

    verb = "updated" if doc is not None else "added"
    # A ``#`` line is a real heading in markdown but a comment in every
    # other format — titles for those come from the file stem.
    if full_path.suffix.lower() in (".md", ".markdown"):
        title = extract_title(content, fallback=full_path.stem)
    else:
        title = full_path.stem
    if doc is None:
        doc = Document(
            source=source,
            path=rel,
            full_path=str(full_path),
            title=title,
            content=content,
            content_hash=digest,
            indexed_at=datetime.now(UTC),
            # Phase 106 (D2/D3): the sourced creation date, normalized
            # (undetermined or future → today). ``created_at_manual``
            # stays the column default (False) — only the date-edit API
            # (task 05) sets it.
            created_at=normalize_doc_date(raw_date),
        )
        session.add(doc)
    else:
        doc.full_path = str(full_path)
        doc.title = title
        doc.content = content
        doc.content_hash = digest
        doc.indexed_at = datetime.now(UTC)
        # Phase 106 (D4): a content change is a new document version —
        # the date is re-sourced and a previous manual correction is
        # reset (it referred to the old content).
        doc.created_at = normalize_doc_date(raw_date)
        doc.created_at_manual = False

    session.flush()  # guarantees doc.id even for brand-new rows

    # Phase 1+2 — chunk, replace the chunk rows (embeddings NULL), embed,
    # and commit the whole file atomically. If the endpoint rejects a chunk
    # as over its input token cap (URL-dense paragraphs tokenize at ~1.1
    # chars/token), halve this file's chunk target and retry — the global
    # policy stays intact for the rest of the KB.
    target = max(400, settings.chunk_target_chars)
    while True:
        chunks_text = chunk_document(content, rel, target, settings.chunk_overlap_chars)
        doc.chunks = [
            Chunk(document_id=doc.id, position=i, content=c) for i, c in enumerate(chunks_text)
        ]
        session.flush()  # delete-orphan cascade drops the previous rows
        if not doc.chunks:
            break
        try:
            vectors = await llm.embed([c.content for c in doc.chunks])
            for row, vec in zip(doc.chunks, vectors, strict=True):
                row.embedding = vec
            break
        except EmbeddingError as e:
            if "token cap" not in str(e) or target <= 400:
                raise
            logger.info(
                "import: re-chunking at %d chars after endpoint token cap: %s",
                target // 2,
                rel,
            )
            target //= 2
    session.commit()

    if verb == "added":
        summary.added += 1
    else:
        summary.updated += 1
    summary.chunks += len(chunks_text)
    logger.info("import: %s source=%s path=%s chunks=%d", verb, source, rel, len(chunks_text))

    # Phase 30; phase 118 (A2, 2026-09-15): EVERY new/changed document
    # gets a ``lite``-model summary — markdown included. Phase 30's
    # "markdown is already natural language" exclusion is retired: the
    # summary is the retrieval seed context (the phase-118 suggestion
    # blocks), not a formatting convenience.
    await _store_summary(
        session, doc=doc, source=source, rel=rel, content=content, llm=llm, summary=summary
    )


async def _store_summary(
    session: Session,
    *,
    doc: Document,
    source: str,
    rel: str,
    content: str,
    llm: Embedder,
    summary: ImportSummary,
    backfill: bool = False,
) -> None:
    """Best-effort ``lite`` summary for one already-committed document.

    Generates the summary (task 03), stores it on ``documents.summary``
    and indexes it as one extra embedded chunk (``is_summary``,
    position −1) that hybrid search can hit instead of badly-formatted
    raw text. Replacement is guaranteed: any pre-existing ``is_summary``
    chunk of this document is deleted first, so at most one summary chunk
    exists per document at a time.

    Best-effort by contract: the doc row + content chunks are committed
    by the caller before this runs, so an :class:`LLMError` /
    :class:`EmbeddingError` only rolls back the summary rows — the file
    stays indexed, without a summary, and the failure is counted in
    ``summary_errors`` (PLAN phase 30).

    ``backfill`` (phase 118, A2): the unchanged-doc NULL-summary path —
    a success counts ``summary_backfilled`` instead of ``summaries``
    (the doc content is untouched, so the import's KB-change signal must
    not move); the rest of the mechanics are identical.

    Image docs (phase 122, task 03, LOCKED A3): for an ``is_image`` doc
    the vision description — ``content`` (which equals ``doc.content``
    on the backfill path) — IS the summary: no ``lite`` call, no
    pointer line (the summary mirrors the description verbatim, so
    ``doc.summary`` == ``doc.content``). The phase-30 chunk mechanics
    (one ``is_summary`` position −1 chunk, replacement, best-effort
    rollback) are unchanged; the only remaining failure class is the
    summary chunk's embed (the ``doc`` row + content chunks survive —
    fail-soft, same as the text path).
    """
    try:
        if doc.is_image:
            # Phase 122 (task 03): the description IS the summary —
            # stored verbatim (no ``lite`` call, no pointer line).
            text = content
        else:
            text = await generate_summary(llm, source=source, path=rel, content=content)
        # Replacement: at most one summary chunk per document at a time.
        # Removing from the collection is what the ``delete-orphan``
        # cascade turns into a row delete on flush — and it keeps the
        # in-memory collection consistent (this session runs with
        # ``expire_on_commit=False``).
        for old in [c for c in doc.chunks if c.is_summary]:
            doc.chunks.remove(old)
        chunk = Chunk(document_id=doc.id, position=-1, content=text, is_summary=True)
        vector = (await llm.embed([text]))[0]
        chunk.embedding = vector
        doc.summary = text
        # Append through the relationship (the ``all`` cascade persists the
        # row) so the collection — live in this session because of
        # ``expire_on_commit=False`` — reflects the committed state.
        doc.chunks.append(chunk)
        session.commit()
        if backfill:
            # Phase 118 (A2): the backfill counts itself apart from fresh
            # imports — the doc content is unchanged, so ``summaries``
            # (a KB-change signal) must not move.
            summary.summary_backfilled += 1
            logger.info(
                "import: summary-backfill source=%s path=%s chars=%d",
                source, rel, len(text),
            )
        else:
            summary.summaries += 1
            logger.info("import: summary source=%s path=%s chars=%d", source, rel, len(text))
    except (LLMError, EmbeddingError) as e:
        session.rollback()
        summary.summary_errors += 1
        logger.error("import: summary failed source=%s path=%s — %s", source, rel, e)


async def _describe_or_skip(
    llm: Embedder, *, data: bytes, source: str, rel: str, full_path: Path
) -> str | None:
    """Phase 122 — the SINGLE seam for the image description (task 03).

    Returns the vision description that becomes the image document's
    ``content`` (and ``summary`` — the ONLY embedded text of the doc),
    or ``None`` when the description failed or came back empty — the
    caller then SKIPS the doc entirely (no row, no copy) and counts
    ``summary.images_failed`` (LOCKED A3, fail-soft: the sync continues,
    the warning + counter are the signal).

    The seam is one line by design (the importer's tests patch exactly
    this function): :func:`app.rag.summarizer.describe_image` — ONE
    CHAT-model (vision) call (LOCKED A3) with the bytes as a data URL
    whose mime comes from :data:`app.rag.summarizer.IMAGE_MIMES`
    (dotted extension; an unlisted ``BOR_IMAGE_EXTENSIONS`` token takes
    the generic fallback — a rejection there fails soft like any other
    description error). ``source``/``rel`` stay on the signature so the
    caller (and the patch) reads like the document being described;
    the failure's doc identity is logged by the caller's warning.
    """
    mime = IMAGE_MIMES.get(full_path.suffix.lower(), IMAGE_FALLBACK_MIME)
    return await describe_image(llm, data=data, mime=mime)


def _delete_image_copy(image_path: str | None) -> None:
    """Best-effort removal of a stale image copy (phase 122).

    A missing path is a no-op (already gone — e.g. the owner cleaned
    the image dir); an unreadable one is logged, never raised — copy
    cleanup must not break the sync (the doc row's fate is decided by
    the upsert/prune logic, not by filesystem hygiene).
    """
    if not image_path:
        return
    try:
        Path(image_path).unlink(missing_ok=True)
    except OSError as e:
        logger.warning("import: could not delete image copy %s — %s", image_path, e)


async def _index_image_file(
    session: Session,
    *,
    source: str,
    rel: str,
    full_path: Path,
    llm: Embedder,
    summary: ImportSummary,
    raw_date: datetime | None = None,
) -> None:
    """The phase-122 image branch of :func:`_index_file` — a standalone
    image is indexed from its BYTES, never its text:

    * the sha256 digest is over the raw bytes (the digest rule is
      content identity — the same bytes are the same document);
    * the PERSISTENT copy lands in ``settings.image_dir`` as
      ``<doc-id>.<ext>`` (the dir is created on demand; the copy is
      written only AFTER a successful description, so a failure never
      leaves an orphan; a changed image deletes the stale copy before
      replacing it);
    * ``content`` is the vision description — the ONLY embedded text of
      the document (the embedding model never sees pixels) — and the
      normal chunk pipeline then embeds it, with the phase-30 summary
      path running on it (the ``is_summary`` position −1 chunk mirrors
      ``Document.summary``).

    ``raw_date`` follows the text path exactly (the phase-106 D2
    fallback: no source date in the map → the file's mtime, read before
    the unchanged early-return because the unchanged path refreshes the
    stored date from the same source; the D1 manual-date lock and the
    D4 refresh apply unmodified).

    Fail-soft (LOCKED A3): a failed/empty description SKIPS the doc
    entirely (no row, no copy) — ``summary.images_failed`` + a warning,
    the sync continues.
    """
    settings = llm.settings
    data = full_path.read_bytes()
    digest = hashlib.sha256(data).hexdigest()
    doc = session.scalar(select(Document).where(Document.source == source, Document.path == rel))
    if raw_date is None:
        # D2 fallback (same as the text path): no source date in the map
        # → the file's mtime (one stat).
        raw_date = file_mtime_datetime(full_path)

    if doc is not None and doc.content_hash == digest:
        # Unchanged image (byte digest) — the text path's unchanged
        # branch, unmodified in shape.
        summary.unchanged += 1
        logger.info("import: unchanged source=%s path=%s", source, rel)
        # Phase 118 (A2) backfill, image flavour: an unchanged image doc
        # whose summary is still NULL (an earlier fail-soft summary miss
        # — for an image, the one remaining failure class: the summary
        # chunk's embed) gets the same best-effort summary pass. For an
        # image the summary IS the stored description (``doc.content``),
        # so the image-aware ``_store_summary`` (task 03) re-stores it
        # verbatim with one ``is_summary`` chunk; a failure (the
        # embed) keeps the doc as-is (no row mutation) — fail-soft,
        # same as the text path.
        if doc.summary is None:
            await _store_summary(
                session, doc=doc, source=source, rel=rel, content=doc.content,
                llm=llm, summary=summary, backfill=True,
            )
        if doc.created_at_manual:
            # D1/D4: the owner's correction survives the sync — no write
            # at all (the text path's manual-date early-return).
            return
        # D4: the date refreshes on every sync, including unchanged
        # files, and may go OLDER (no monotonic guard).
        target = normalize_doc_date(raw_date)
        if target != doc.created_at:
            doc.created_at = target
            session.commit()
            summary.dates_updated += 1
            logger.info(
                "import: date-refreshed source=%s path=%s date=%s",
                source, rel, doc.created_at.isoformat(),
            )
        return

    verb = "updated" if doc is not None else "added"
    # The description is the doc's content (task 03: and its summary) —
    # it is generated BEFORE anything is written, so a failure skips the
    # doc with no row and no copy (the copy is only made after a
    # successful description — a failure never leaves an orphan).
    content = await _describe_or_skip(
        llm, data=data, source=source, rel=rel, full_path=full_path
    )
    if content is None:
        # LOCKED A3 fail-soft: an undescribed image is unsearchable
        # noise — skip the doc entirely (no row, no copy).
        summary.images_failed += 1
        logger.warning("import: image description failed source=%s path=%s", source, rel)
        return

    # The persistent copy: uploads are replaced on every upload, git
    # checkouts are re-cloned, local dirs are user-edited — the served
    # bytes must outlive the source file. Named by the doc id: a new
    # doc's id is the uuid4 chosen here (row and copy agree); a changed
    # doc keeps its id (the copy path is stable).
    doc_id = doc.id if doc is not None else uuid.uuid4()
    image_dir = Path(settings.image_dir).expanduser()
    image_dir.mkdir(parents=True, exist_ok=True)
    if doc is not None:
        # A CHANGED image (hash differs): the stale copy is deleted
        # before replacement.
        _delete_image_copy(doc.image_path)
    copy_path = image_dir / f"{doc_id}{full_path.suffix.lower()}"
    copy_path.write_bytes(data)

    # The non-markdown title rule (the image's content is prose, but the
    # doc IS the image — the file stem is the title).
    title = full_path.stem
    if doc is None:
        doc = Document(
            id=doc_id,
            source=source,
            path=rel,
            full_path=str(full_path),
            title=title,
            content=content,
            content_hash=digest,
            indexed_at=datetime.now(UTC),
            created_at=normalize_doc_date(raw_date),
            is_image=True,
            image_path=str(copy_path),
        )
        session.add(doc)
    else:
        doc.full_path = str(full_path)
        doc.title = title
        doc.content = content
        doc.content_hash = digest
        doc.indexed_at = datetime.now(UTC)
        # Phase 106 (D4): a content change is a new document version —
        # the date is re-sourced and a previous manual correction is
        # reset (it referred to the old content).
        doc.created_at = normalize_doc_date(raw_date)
        doc.created_at_manual = False
        doc.is_image = True
        doc.image_path = str(copy_path)

    session.flush()  # guarantees doc.id even for brand-new rows

    # Phase 1+2 — the UNCHANGED pipeline on the description: chunk,
    # replace the chunk rows (embeddings NULL), embed, and commit the
    # whole file atomically (one transaction per file). The token-cap
    # retry loop is copied from the text path; a description is short,
    # so it never fires in practice.
    target = max(400, settings.chunk_target_chars)
    while True:
        chunks_text = chunk_document(content, rel, target, settings.chunk_overlap_chars)
        doc.chunks = [
            Chunk(document_id=doc.id, position=i, content=c) for i, c in enumerate(chunks_text)
        ]
        session.flush()  # delete-orphan cascade drops the previous rows
        if not doc.chunks:
            break
        try:
            vectors = await llm.embed([c.content for c in doc.chunks])
            for row, vec in zip(doc.chunks, vectors, strict=True):
                row.embedding = vec
            break
        except EmbeddingError as e:
            if "token cap" not in str(e) or target <= 400:
                raise
            logger.info(
                "import: re-chunking at %d chars after endpoint token cap: %s",
                target // 2,
                rel,
            )
            target //= 2
    session.commit()

    if verb == "added":
        summary.added += 1
    else:
        summary.updated += 1
    summary.chunks += len(chunks_text)
    logger.info("import: %s source=%s path=%s chunks=%d", verb, source, rel, len(chunks_text))

    # Phase 30 shape on the description — image-aware (task 03, LOCKED
    # A3): the description IS the summary (stored verbatim, no
    # ``lite`` call), so ``doc.summary`` == ``doc.content`` and the
    # ``is_summary`` position −1 chunk mirrors it.
    await _store_summary(
        session, doc=doc, source=source, rel=rel, content=content, llm=llm, summary=summary
    )


def _prune(
    session: Session,
    source_names: set[str],
    seen: set[tuple[str, str]],
    images: bool = False,
) -> int:
    """Delete documents of *source_names* whose file is no longer in *seen*.

    *images* (phase 122 prune guard, LOCKED, derived from A3/A4): while
    the image toggle is OFF (``images`` False), every ``is_image`` doc is
    SKIPPED — the image is invisible to an images-off walk, not a deleted
    file, so pruning it would silently destroy image documents on the
    first images-off sync. Toggle ON → normal semantics: a deleted image
    file prunes its doc, and the pruned image's ``image_dir`` copy is
    deleted with it.
    """
    if not source_names:
        return 0
    pruned = 0
    docs = session.scalars(select(Document).where(Document.source.in_(source_names))).all()
    for doc in docs:
        if (doc.source, doc.path) not in seen:
            if doc.is_image and not images:
                # Prune guard: invisible to the walk, not deleted.
                continue
            _delete_image_copy(doc.image_path)
            session.delete(doc)
            pruned += 1
            logger.info("import: pruned source=%s path=%s", doc.source, doc.path)
    if pruned:
        session.commit()
    return pruned