phase: 129_sync_chunk_progress

All green. Final report:

**Phase 129 final verification pass — 100% green** (previous work intact; this pass re-verified everything and fixed two pre-existing README inconsistencies)

- Verified on-spec: `LLMClient.embed` per-batch `on_progress` hook; importer `on_chunk_progress` plumbing (text/image/PDF branches; summary embed excluded); `SyncStatus.current_chunk/chunks_total` reset+clear contract; UI `.sync-label-chunk` pinned part with untruncated `title`/`#sync-result`
- `uv run pytest --cov=app -q`: exit 0, ~3041 passed, 0 failed, `app/` coverage **99%** (>90% gate)
- `uv run pytest tests/e2e/test_sync_chunk_progress.py -v --no-cov` (isolated, DB up): **2 passed** (endpoint chunk climb with true `chunk_document` total + UI `· chunk n/m` part)
- `uv run ruff check .`: all checks passed · `uv run pyright`: 0 errors, 0 warnings
- Completion criteria: all 5 ✅ (incl. no behavior change in completed phases — full suite green, CLI/chat no-op paths byte-identical)
- Docs: README.md fixed 2 pre-existing inconsistencies (upload section claimed "unpacks + scans immediately" — phase 90 made it unpack+register only, scan deferred to Sync; troubleshooting claimed "120s client timeout" — actual guard is the 300 s pre-token `TURN_TIMEOUT_MS`). Phase-129 label behavior already documented. AGENTS.md re-verified consistent — no changes
- Removed stray untracked `NOTES.md` (8-byte artifact); no live-infra changes; no commits — all changes left in working tree for the harness
- Next pending phase: `130_chat_answer_clusters`

tests/unit/test_importer.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import os
+from collections.abc import Callable
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -27,6 +28,7 @@ from sqlalchemy.orm import Session
 import app.rag.importer as importer
 from app.config import Settings
 from app.models import Chunk, Document
+from app.rag.chunker import chunk_document
 from app.rag.importer import (
     EXCLUDED_DIRS,
     ImportSummary,
@@ -48,10 +50,15 @@ DEFAULT_EXTS = frozenset({".md", ".markdown", ".txt", ".yaml", ".yml", ".json",
 class _PoisonEmbedder(FakeEmbedder):
     """Fails (like a real endpoint) on any text containing 'poison'."""
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
+    async def embed(
+        self,
+        texts: list[str],
+        *,
+        on_progress: Callable[[int, int], None] | None = None,
+    ) -> list[list[float]]:
         if any("poison" in t for t in texts):
             raise EmbeddingError("embeddings endpoint refused the input (simulated)")
-        return await super().embed(texts)
+        return await super().embed(texts, on_progress=on_progress)
 
 
 class _FailingChatEmbedder(FakeEmbedder):
@@ -67,13 +74,18 @@ class _CapEmbedder(FakeEmbedder):
     """Simulates the endpoint's ~1024-token input cap at ~1.1 chars/token:
     any single text over 1000 chars is rejected (URL-dense worst case)."""
 
-    async def embed(self, texts: list[str]) -> list[list[float]]:
+    async def embed(
+        self,
+        texts: list[str],
+        *,
+        on_progress: Callable[[int, int], None] | None = None,
+    ) -> list[list[float]]:
         if any(len(t) > 1000 for t in texts):
             raise EmbeddingError(
                 "a single 1100-char chunk exceeded the endpoint's per-request "
                 "input token cap — lower BOR_CHUNK_TARGET_CHARS and re-import"
             )
-        return await super().embed(texts)
+        return await super().embed(texts, on_progress=on_progress)
 
 
 def _embedder_with_extensions(extensions: str) -> FakeEmbedder:
@@ -1430,3 +1442,205 @@ def test_prune_records_changed_paths_only_when_summary_given(db, tmp_path: Path)
         assert counters.changed_paths == [(source, "gone2.md")]
     finally:
         _cleanup_source(db, root.name)
+
+
+# ---------- phase 129 (task 02): on_chunk_progress plumbing ----------
+
+
+class _BatchingEmbedder(FakeEmbedder):
+    """Mimics ``LLMClient``'s top-level token-budgeted batching (the
+    task-01 contract on the importer side): *texts* is split into
+    consecutive groups of *batch_size* and the hook fires
+    ``(first_chunk_1based, total)`` before EACH group — the group is then
+    embedded as one request (``super().embed`` with ``on_progress=None``:
+    this class owns the hook sequence, so the base's single-batch
+    announce is suppressed per group)."""
+
+    def __init__(self, batch_size: int) -> None:
+        super().__init__()
+        self.batch_size = batch_size
+
+    async def embed(
+        self,
+        texts: list[str],
+        *,
+        on_progress: Callable[[int, int], None] | None = None,
+    ) -> list[list[float]]:
+        out: list[list[float]] = []
+        for start in range(0, len(texts), self.batch_size):
+            group = texts[start : start + self.batch_size]
+            if on_progress is not None:
+                on_progress(start + 1, len(texts))
+            out.extend(await super().embed(group, on_progress=None))
+        return out
+
+
+def _markdown_with_sections(stem: str, n: int) -> str:
+    """One markdown file with a preamble (h1 title + intro) and *n* short
+    ``##`` sections — each section is under the 2000-char target, so the
+    file chunks to exactly *n* + 1 content chunks (the preamble + one per
+    section; verified against ``chunk_document`` below)."""
+    lines = [f"# {stem}", "", f"Intro line for {stem.lower()}.", ""]
+    for i in range(1, n + 1):
+        lines += [f"## {stem} {i}", "", f"Body text for section {i}.", ""]
+    return "\n".join(lines)
+
+
+def _chunk_count(content: str, rel: str) -> int:
+    """The content-chunk count the importer's pipeline will produce for
+    *content* (the same ``chunk_document`` call and defaults as
+    ``_replace_content_chunks``: ``max(400, chunk_target_chars)`` target,
+    ``chunk_overlap_chars`` overlap)."""
+    s = Settings(_env_file=None)  # pyright: ignore[reportCallIssue]
+    return len(chunk_document(content, rel, max(400, s.chunk_target_chars), s.chunk_overlap_chars))
+
+
+def test_on_chunk_progress_reports_per_file_batch_sequences(
+    db, tmp_path: Path
+) -> None:
+    """A multi-batch file's embed position is fully observable: the hook
+    fires before each embed batch with ``(first_chunk_1based, total)`` —
+    ``total`` is the file's content-chunk count, the first indices climb
+    by the previous batch's size, and a SECOND file in the same run gets
+    its OWN sequence (the two files' totals never mix)."""
+    root = tmp_path / "chunkprog"
+    root.mkdir()
+    content_a = _markdown_with_sections("Alpha", 8)  # preamble + 8 = 9 chunks
+    content_b = _markdown_with_sections("Beta", 4)  # preamble + 4 = 5 chunks
+    (root / "alpha.md").write_text(content_a)
+    (root / "beta.md").write_text(content_b)
+    n_a, n_b = _chunk_count(content_a, "alpha.md"), _chunk_count(content_b, "beta.md")
+    assert (n_a, n_b) == (9, 5)
+
+    batch_size = 3
+
+    def expected(n: int) -> list[tuple[int, int]]:
+        return [(start + 1, n) for start in range(0, n, batch_size)]
+
+    # The task's multi-batch case: file A spans >= 2 batches.
+    assert len(expected(n_a)) >= 2
+
+    llm = _BatchingEmbedder(batch_size=batch_size)
+    chunk_events: list[tuple[int, int]] = []
+    file_events: list[tuple[str, str, int, int]] = []
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                progress=lambda s, r, d, t: file_events.append((s, r, d, t)),
+                on_chunk_progress=lambda f, t: chunk_events.append((f, t)),
+            )
+        )
+        assert (summary.added, summary.chunks) == (2, n_a + n_b)
+    finally:
+        _cleanup_source(db, root.name)
+
+    # The per-file ``progress`` hook names each file in walk order ...
+    assert file_events == [
+        (root.name, "alpha.md", 1, 2),
+        (root.name, "beta.md", 2, 2),
+    ]
+    # ... and the chunk sequence is exactly file A's batches followed by
+    # file B's OWN restarted sequence (its totals never leak into A's).
+    assert chunk_events == expected(n_a) + expected(n_b)
+
+
+def test_on_chunk_progress_single_batch_file_and_summary_excluded(
+    db, tmp_path: Path
+) -> None:
+    """A file whose chunks fit in ONE embed batch reports exactly one
+    ``(1, N)`` event, and the document's summary-chunk embed (the SECOND
+    ``llm.embed`` call) NEVER fires the hook — the counter counts content
+    chunks only (A6: no ``1/1`` noise after the real progress ends)."""
+    root = tmp_path / "chunkprog-one"
+    root.mkdir()
+    content = _markdown_with_sections("Gamma", 3)  # preamble + 3 = 4 chunks
+    (root / "gamma.md").write_text(content)
+    n = _chunk_count(content, "gamma.md")
+    assert n == 4
+    events: list[tuple[int, int]] = []
+    llm = FakeEmbedder()
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_chunk_progress=lambda f, t: events.append((f, t)),
+            )
+        )
+        assert summary.added == 1
+    finally:
+        _cleanup_source(db, root.name)
+    assert events == [(1, n)]
+    # Two embed calls happened (the content batch + the excluded summary
+    # chunk) — the hook fired for the first only.
+    assert [len(c) for c in llm.calls] == [n, 1]
+
+
+def test_on_chunk_progress_image_branch_reports_one_of_one(
+    db, tmp_path: Path
+) -> None:
+    """The image branch's ONE description-chunk embed reports exactly
+    ``(1, 1)`` — it IS the file's only chunk — and the mirrored
+    summary-chunk embed never fires the hook. The REAL description seam
+    is used (the ``_VisionLLM`` multimodal ``chat``)."""
+    root = tmp_path / "imgchunk"
+    root.mkdir()
+    (root / "pic.png").write_bytes(_png_bytes(4, 4, (0.8, 0.2, 0.2)))
+    events: list[tuple[int, int]] = []
+    llm = _VisionLLM()
+    llm.settings = Settings(
+        _env_file=None,  # pyright: ignore[reportCallIssue]
+        images=True,
+        image_dir=str(tmp_path / "images"),
+    )
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_chunk_progress=lambda f, t: events.append((f, t)),
+            )
+        )
+        assert (summary.files, summary.added) == (1, 1)
+    finally:
+        _cleanup_source(db, root.name)
+    assert events == [(1, 1)]
+    # The content embed (the 1 description chunk) + the excluded
+    # summary-chunk embed — the hook fired for the first only.
+    assert [len(c) for c in llm.calls] == [1, 1]
+
+
+def test_on_chunk_progress_pdf_branch_reports_content_chunks(
+    db, tmp_path: Path
+) -> None:
+    """The PDF branch's CONTENT embed is the primary target: a two-page
+    PDF (two ``Page N:`` chunks) reports ``(1, 2)`` once through the
+    hook, and the summary-chunk embed never fires it."""
+    root = tmp_path / "pdfchunk"
+    root.mkdir()
+    doc = fitz.open()
+    doc.new_page().insert_text((72, 72), "First page text about networking.")
+    doc.new_page().insert_text((72, 72), "Second page text about storage.")
+    doc.save(root / "report.pdf")
+    doc.close()
+    events: list[tuple[int, int]] = []
+    llm = _VisionLLM()
+    llm.settings = Settings(
+        _env_file=None,  # pyright: ignore[reportCallIssue]
+        images=True,
+        image_dir=str(tmp_path / "images"),
+        import_extensions="md,txt,pdf",
+    )
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_chunk_progress=lambda f, t: events.append((f, t)),
+            )
+        )
+        assert (summary.files, summary.added) == (1, 1)
+    finally:
+        _cleanup_source(db, root.name)
+    assert events == [(1, 2)]
+    # The content embed (the 2 page chunks, one batch) + the excluded
+    # summary-chunk embed — the hook fired for the first only.
+    assert [len(c) for c in llm.calls] == [2, 1]