phase: 152_sync_progress_image_pass

All criteria verified. The CLI path is hook-less (only `app/api/sync.py` passes the hook), and the byte-identity is pinned by `test_on_image_progress_none_run_is_byte_identical`.

## Final Report — Phase 152 final verification pass

- **Verified (no new code needed):** `SyncStatus.images_done/images_total` + reset/clear rules in `app/api/sync.py`; `on_image_progress` threaded through the importer's three PDF vision spans with the continuous per-file counter; the `_image_hook` closure; `sources.js` `· image n/m` label part (box + title + aria).
- **C1** status climbs 1→N + UI shows `· image n/m`: ✅ unit 3/15→15/15, integration 1/15→7/15, E2E 1/4→4/4 (title/aria ride the full string)
- **C2** never frozen: ✅ E2E asserts ≥3 distinct positions, starting at 1, non-decreasing, ≤ total, reaching it
- **C3** cleared at overview/summaries + all terminals; per-file reset via `_hook`; no stale position for image-less files: ✅ unit pins + E2E settle (image part gone in box/title/aria, endpoint 0/0)
- **C4** `/api/sync/status` additive-only: ✅ full-dict pin passes (all pre-existing keys unchanged + the two new 0/0 keys)
- **C5** hook-less caller byte-identical: ✅ CLI passes no hook; `test_on_image_progress_none_run_is_byte_identical` green
- **C6** `uv run pytest`: **3526 passed, exit 0** (95s); dedicated E2E in isolation: **2 passed** (95s); `uv run ruff check . && uv run pyright`: **clean**; coverage: **99% on `app/` (>90% gate)**
- Regression: all five neighbouring E2E suites green in isolation (progress_box 8, chunk_progress 2, label_mobile 4, pdf_support 6, image_documents 4)
- **C7** atomic commit: left to the harness (no git index/history touched per rules)
- **Docs:** README.md — sync-label paragraph already updated this phase, accurately; all `.env.example` vars documented; routes/commands verified — **no changes needed**. AGENTS.md — all references (PLAN, user_stories, test_smoke.py, .env.example, gpg config, .gitignore) verified — **no changes needed**
- **Defects found in prior work:** none (one prior env-leak defect was already fixed inside this phase's task 04, conftest)
- No live-infrastructure touched; no ntfy used. **Next pending phase:** none — `todo/` held only this phase

tests/conftest.py

@@ -48,6 +48,12 @@ from app.config import Settings as _Settings  # noqa: E402
 
 os.environ["BOR_DOCS_REPO"] = ""
 os.environ["BOR_SUGGESTIONS"] = json.dumps(_Settings.model_fields["suggestions"].default)
+# Phase 152 (task 04, defect fix): the same leak class for the display
+# name — an operator's local ``.env`` may legitimately carry
+# ``BOR_APP_NAME`` (the dev environment runs a renamed build), and the
+# default-metadata pins (``/api/config`` ``app_name``, the page titles
+# + header brand) must see the code default.
+os.environ["BOR_APP_NAME"] = _Settings.model_fields["app_name"].default
 
 # Phase 62: the same leak class for the new UI customization settings —
 # an operator's local ``.env`` may legitimately carry

tests/e2e/test_sync_image_progress.py

@@ -0,0 +1,885 @@
+"""Phase 152 task 05 E2E (Playwright, mock-only): during a PDF sync whose
+images take time to describe, the live label shows the image position
+MOVING — the owner's "stuck for several minutes" (TODO.md:3: the label
+sat on ``Syncing… … (2540/2540) · chunk 125/126`` through the PDF's
+vision spans) is impossible on the fixture.
+
+The dedicated story suite for ``152_sync_progress_image_pass`` (A16 —
+one Playwright file per phase, run in isolation): the ``on_image_progress
+hook (phase 152, tasks 02/03) fires immediately before each vision
+describe call with ``(position, total)``, the runner's ``_image_hook``
+closure lands it on ``SyncStatus.images_done/images_total``, and
+``fmtSyncPhaseLabel`` renders ``· image n/m`` — the phase-149 box
+(``#sync-progress``) is the surface that shows it, the button
+``title``/``aria-label`` carry the UNTRUNCATED composed string (A4).
+
+Run in isolation (DB must be up: ``podman compose up -d db``):
+
+    uv run pytest tests/e2e/test_sync_image_progress.py -v --no-cov
+
+MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
+deterministic mock plus the phase-64 ``tests/e2e/slow_llm.py`` delay
+proxy in front of it (``SLOW_DELAY_S = 1.5 s``) that stretches EVERY
+LLM call (vision describes included) so the fixture's vision span
+outlives the UI's 2 s status poll and each position outlives the
+recorder's ~300 ms page-sampling cadence by ~10× — the observed
+positions are dense samples of a real climb, not races.
+
+Timing fixture (the phase-129/98 rationale, re-sized for vision spans):
+the fixture's ONE-PDF run is ≈ 27 LLM calls ≈ 40 s: 2 model probes +
+10 in-flight embed batches (the ~4 KB-per-page text packs into 20
+content chunks at the app's code-default, pinned sizing — the token
+budget groups them into 10 requests of 2) + 4 image pipelines (one
+vision describe + 2 content-embed batches each — the image doc's
+context + description content packs into 3 chunks) + 1 PDF summary +
+1 overview + 1 folder summary. The image span itself is 4 × (1.5 s
+describe + 2 × 1.5 s embed) = 18 s: each of the 4 positions holds
+for 4.5 s — that outlives the UI's 2 s poll (at least one tick lands
+in every 4.5 s window — every position is rendered, not a race) and
+the test's ~300 ms page sampling sees every rendered value.
+
+KB fixture (the ``test_local_directory_sources.py`` / phase-94 API
+registration pattern — a host temp dir, the app runs on the same host;
+no git anywhere): ``imgpass/`` with EXACTLY ONE file, a PyMuPDF-built
+``imgpass.pdf`` (LOCKED A1) — 10 text pages (each ~4 KB of
+deterministic filler, well over the 100-char bar — the phase-140
+pictorial gate never fires, and the layer is clean — the phase-133
+fallback never fires: the EMBEDDED-IMAGE pass is the file's one and
+only vision span) + 4 GENUINELY DISTINCT non-blank images on pages
+1–4 (different sizes + seeds → different stored bytes — the phase-151
+dedupe keeps all four; colorful noise, sample mean ~127 — well under
+the 235 near-white floor, so the blank gate keeps all four: the
+span's total is the post-dedupe/post-gate count, the work actually
+about to happen). The expected chunk count is computed IN THE TEST by
+calling ``extract_pdf_text`` (the SAME parser the importer uses) and
+``chunk_document`` (the SAME function + arguments — the importer's
+``max(400, target)`` floor + the overlap setting — as
+``_replace_content_chunks``), read back from the PDF on disk — the
+ground truth never drifts from the app under test. The sizing knobs
+are pinned to the code defaults in the app-under-test env (the
+leak-guard pattern — an operator's gitignored ``.env`` tuning
+``BOR_CHUNK_TARGET_CHARS`` must not change the suite's ground truth).
+
+The suite runs ONE module app (it binds its own port — the conftest
+session app is never started in an isolated run) over the shared E2E
+Postgres: the SLOW app (``BOR_LLM_BASE_URL`` = the slow proxy in front
+of the mock), ``BOR_IMAGES=true`` + the ``pdf`` import token pinned
+EXPLICITLY (an operator's local gitignored ``.env`` cannot leak the
+scope in either direction) + ``BOR_PDF_PICTORIAL`` pinned off (the
+house E2E convention — this suite's span is the embedded pass; the
+fixture's ≥100-char pages make the pictorial pass a no-op even if the
+knob leaked on).
+
+Autouse cleanup (the phase-96 pattern): before each test wait for no
+running sync on the module app (the run state lives in its memory),
+then truncate the shared registry + KB tables (documents/chunks/
+query_log/steering_notes/kb_overview/git_sources/sources_meta/
+folder_summaries) — each test seeds its own source row and runs its
+own sync; after each test the same truncate.
+
+Test → observable mapping (Playwright Mapping Rule):
+1. ``test_label_advances_through_the_image_span`` (the anti-freeze
+   proof — the UI's own 2 s poll, densely sampled at ~300 ms): after
+   clicking **Sync sources**, the test collects the DISTINCT
+   ``#sync-progress-text`` values seen while the run is in flight
+   (first-seen order) — and asserts: (a) at least THREE distinct
+   labels containing ``· image `` are observed (the freeze the owner
+   reported is a single repeated label — "≥3 distinct positions" IS
+   the bug's negation), the positions START at 1, are
+   NON-DECREASING in observation order, never EXCEED the total (4 —
+   the post-gate count), and REACH it; (b) the CHUNK part appeared
+   EARLIER in the run (the first ``· chunk n/20`` label precedes the
+   first ``· image `` label — the two positions are distinct) and
+   every image-span label carries the chunk part WITH IT (the last
+   chunk position persists through the file's image span —
+   "… (1/1) · chunk 19/20 · image 3/4"), its total == the file's
+   TRUE chunk count; (c) the button ``title`` + ``aria-label`` carry
+   the SAME full string as the box on every running sample (A4 —
+   asserted per sample, atomically); (d) at settle the box shows the
+   counts line (``5 added`` — the PDF + its 4 image docs), the label
+   reads the terminal ``Synced HH:MM``, the count + chunk parts are
+   hidden + empty, and the image part is GONE (no stale position in a
+   terminal state — the endpoint's terminal carries 0/0 + the cleared
+   chunk position while the file counts keep their final values).
+2. ``test_mid_run_reload_re_attaches_with_image_position`` (the
+   never-stale contract): the API starts the sync, the page's own
+   poll renders the image span (the FIRST observed position is
+   ``1/4`` — the whole span is ahead), the test reloads the page
+   mid-run, and the load-time re-attach (``initSyncButton`` — a
+   running sync re-enters with the WHOLE status, phase 98) renders
+   the CURRENT image position with the A4 title/aria — then the run
+   settles to the counts line.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pymupdf as fitz  # PyMuPDF — the fixture-PDF builder (LOCKED A1)
+import pytest
+from playwright.sync_api import Page, expect
+from sqlalchemy import text
+
+from app.config import Settings as _Settings
+from app.db import SessionLocal
+from app.rag.chunker import chunk_document
+from app.rag.llm import LLMClient
+from app.rag.pdf import extract_pdf_text, is_usable_pdf_text
+from e2e.auth_helpers import login
+from e2e.conftest import (
+    ADMIN_PASSWORD,
+    MOCK_PORT,
+    SESSION_SECRET,
+    USE_REAL_LLM,
+    _wait_http,
+)
+
+REPO = Path(__file__).resolve().parents[2]
+
+# Phase 79 (task 04, full inventory): the conftest session app owns its
+# port in a combined run — this module's app binds its own port
+# instead (a same-port second uvicorn dies on bind and would drive the
+# wrong server). Env-overridable.
+APP_PORT = int(os.environ.get("E2E_APP_PORT_IMGPROG", "8171"))
+APP_URL = f"http://127.0.0.1:{APP_PORT}"
+
+#: The slow-LLM proxy's port (the conftest's mock LLM stays on
+#: MOCK_PORT — the proxy fronts it).
+SLOW_PORT = int(os.environ.get("E2E_SLOW_LLM_PORT_IMGPROG", "8909"))
+SLOW_URL = f"http://127.0.0.1:{SLOW_PORT}"
+
+#: Per-LLM-request delay on the proxy — sized so each in-flight VISION
+#: PIPELINE's position (the describe + the content embed) outlives the
+#: UI's 2 s poll (see the module docstring's timing fixture paragraph).
+SLOW_DELAY_S = "1.5"
+
+# --------------------------------------------------------------------------
+# Fixture constants (deterministic, token-controlled)
+# --------------------------------------------------------------------------
+
+#: The local source — the temp directory's basename (``kind=local`` →
+#: the directory's basename is the source name, phase 38).
+SOURCE = "imgpass"
+#: The single fixture file (the source is one file — the count part
+#: reads "(1/1)" throughout the import phase).
+PDF_NAME = "imgpass.pdf"
+N_FILES = 1
+
+#: The fixture PDF's shape — 10 text pages (each ~4 KB of filler —
+#: 20 content chunks at the code-default sizing, 10 embed batches) +
+#: 4 distinct non-blank images on pages 1–4 (the span's total —
+#: post-dedupe, post-blank-gate: the work actually about to happen).
+N_PAGES = 10
+N_IMAGES = 4
+
+#: The settled counts line (phase 150 — the box is the sole sync text
+#: surface): the PDF + its 4 embedded-image docs, a fresh KB (nothing
+#: updated/unchanged/pruned).
+SETTLE_TEXT = f"{1 + N_IMAGES} added"
+
+#: The three sizing knobs, pinned to the CODE DEFAULTS in both the
+#: app-under-test env (the leak guards below) and the test process's
+#: expected-value computation — an operator's gitignored ``.env``
+#: (which may legitimately tune ``BOR_CHUNK_TARGET_CHARS``) cannot
+#: change the suite's ground truth. Derived from the class fields, so
+#: they never drift from app/config.py (the conftest leak-guard
+#: pattern).
+CHUNK_TARGET = _Settings.model_fields["chunk_target_chars"].default
+CHUNK_OVERLAP = _Settings.model_fields["chunk_overlap_chars"].default
+EMBED_BATCH_SIZE = _Settings.model_fields["embed_batch_size"].default
+
+#: The test process's mirror of the app-under-test sizing settings
+#: (explicit kwargs beat any ``.env`` for the fields the expected
+#: values depend on).
+_EXPECTED_SETTINGS = _Settings(
+    chunk_target_chars=CHUNK_TARGET,
+    chunk_overlap_chars=CHUNK_OVERLAP,
+    embed_batch_size=EMBED_BATCH_SIZE,
+)
+
+#: "Synced HH:MM" — the local-time last-result label (sources.js's
+#: fmtSyncTime), any hour/minute (the test_sync_button.py pattern).
+SYNCED_LABEL = re.compile(r"Synced \d{1,2}:\d{2}$")
+#: The settled boot label — idle ("Sync sources") or the RE-ATTACHED
+#: terminal from the previous test's sync (the module app is shared,
+#: its in-memory sync state survives the test boundary, and the
+#: phase-32 boot re-attach renders the last result).
+IDLE_OR_SETTLED = re.compile(r"^(Sync sources|Synced \d{1,2}:\d{2})$")
+#: The chunk part (the phase-129 unit-pinned shape): "· chunk n/m" —
+#: ``m`` is asserted against the expected chunk count by the test.
+CHUNK_PART_RE = re.compile(r"· chunk (\d+)/(\d+)")
+#: The image part (the phase-152 unit-pinned shape): "· image n/m".
+IMAGE_PART_RE = re.compile(r"· image (\d+)/(\d+)")
+
+#: Generous settle budgets: a changed sync against the slowed LLM is
+#: ≈ 40 s; the UI's 2 s poll settles at most one tick after the
+#: terminal state lands.
+SETTLE_TIMEOUT_MS = 120_000
+
+
+# ---------------------------------------------------------------------------
+# The fixture PDF (built with PyMuPDF — LOCKED A1)
+# ---------------------------------------------------------------------------
+
+
+def _noise_png(w: int, h: int, seed: int) -> bytes:
+    """A deterministic colorful-noise PNG (the "real image" stand-in).
+    Different ``(w, h, seed)`` → different stored bytes (the phase-151
+    byte-dedupe keeps all four); the formula's sample mean is ~127 —
+    well under the phase-151 blank gate's 235 near-white floor, and
+    the samples scatter far outside the ±17 band — every image
+    survives the gate with room to spare (see the module docstring's
+    KB fixture paragraph)."""
+    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, w, h))
+    for y in range(h):
+        for x in range(w):
+            pix.set_pixel(
+                x,
+                y,
+                ((x * 7 + y * 11 + seed * 13) % 256,
+                 (x * 3 + y * 17 + seed * 29) % 256,
+                 (x * 5 + y * 23 + seed * 41) % 256),
+            )
+    return pix.tobytes("png")
+
+
+#: The four fixture images' shapes (distinct sizes — the distinct-
+#: bytes guarantee even if two seeds collided) — one per page 1–4.
+_IMAGE_SHAPES = [(24, 18), (32, 24), (40, 20), (28, 28)]
+
+
+def _page_text(k: int) -> str:
+    """One page's deterministic ~4 KB filler (no randomness — the
+    expected chunk count is stable run to run)."""
+    base = (
+        f"Page {k} notes for the imgpass fixture document carry "
+        f"deterministic filler number {k} of the ten-page pdf, written "
+        f"out to size so the chunker has real content to work with "
+        f"across the page. "
+    )
+    return base * 11
+
+
+def _build_pdf(pdf_path: Path) -> None:
+    """The fixture PDF: ``N_PAGES`` text pages (``insert_textbox`` —
+    the return value is checked — an overfull rect would silently
+    drop text and skew the ground truth) + the 4 images on pages 1–4
+    (distinct ``insert_image`` streams — distinct xrefs, distinct
+    bytes). The metadata title is the stem (no title — the stem
+    fallback, phase 126 A2)."""
+    doc = fitz.open()
+    for k in range(1, N_PAGES + 1):
+        page = doc.new_page()
+        rc = page.insert_textbox(
+            fitz.Rect(48, 90, 548, 812), _page_text(k), fontsize=10
+        )
+        assert rc >= 0, f"the page {k} text did not fit the rect (rc={rc})"
+        if k <= N_IMAGES:
+            w, h = _IMAGE_SHAPES[k - 1]
+            page.insert_image(
+                fitz.Rect(72, 820 - h * 2, 72 + w * 2, 860),
+                stream=_noise_png(w, h, k),
+            )
+    doc.save(pdf_path)
+    doc.close()
+
+
+def _expected_chunks(pdf_path: Path) -> list[str]:
+    """The PDF's expected content chunks — the SAME function and
+    arguments the importer's ``_replace_content_chunks`` uses
+    (``chunk_document(content, rel, max(400, target), overlap)`` — the
+    importer's token-cap floor rule; ``rel`` = the source-relative
+    path) over the SAME parser the importer reads
+    (``extract_pdf_text`` — the ``---PAGE-N---`` page markers): the
+    endpoint's ``chunks_total`` must equal
+    ``len(chunk_document(...))`` — the file's TRUE chunk count, not a
+    batch count."""
+    content = extract_pdf_text(pdf_path)
+    return chunk_document(
+        content, PDF_NAME, max(400, CHUNK_TARGET), CHUNK_OVERLAP
+    )
+
+
+def _expected_batch_count(chunks: list[str]) -> int:
+    """The expected embed-batch count — the SAME
+    ``LLMClient._batch_texts`` partition the app's embed loop walks
+    (the phase-129 unit pattern — call it directly for the boundaries).
+    The task's sizing bar: ≥ 2 batches (the chunk part's climb is a
+    real sequence, not a single point)."""
+    return len(LLMClient(_EXPECTED_SETTINGS)._batch_texts(chunks))
+
+
+# ---------------------------------------------------------------------------
+# The in-browser snapshot functions (Playwright Mapping Rule — the
+# rendered DOM is the observable).
+#
+# ``_JS_SAMPLE`` — test 1's dense sampler: the box text + the button
+# title/aria read in the SAME in-browser evaluation (a 2 s poll tick
+# updates the parts + the title together in one JS task, so the atomic
+# check can never straddle one — the phase-128 pattern); ``running``
+# marks the in-flight labels (the A4 title/aria contract applies to
+# them).
+#
+# ``_JS_IMAGE_SNAP`` — test 2's wait: null until the box carries the
+# ``· image n/<arg>`` part, then the texts (read atomically with the
+# title/aria).
+# ---------------------------------------------------------------------------
+
+_JS_SAMPLE = """
+() => {
+    const boxTextEl = document.querySelector('#sync-progress-text');
+    const btn = document.querySelector('#sync-btn');
+    if (!boxTextEl || !btn) return null;
+    const box = boxTextEl.textContent;
+    return {
+        box: box,
+        title: btn.getAttribute('title') || '',
+        aria: btn.getAttribute('aria-label') || '',
+        running: box.startsWith('Syncing…'),
+    };
+}
+"""
+
+_JS_IMAGE_SNAP = """
+(n) => {
+    const boxTextEl = document.querySelector('#sync-progress-text');
+    const btn = document.querySelector('#sync-btn');
+    if (!boxTextEl || !btn) return null;
+    const box = boxTextEl.textContent;
+    const m = box.match(/^Syncing… .*· image (\\d+)\\/(\\d+)$/);
+    if (!m || Number(m[2]) !== n) return null;
+    return {
+        box: box,
+        posN: Number(m[1]),
+        title: btn.getAttribute('title') || '',
+        aria: btn.getAttribute('aria-label') || '',
+    };
+}
+"""
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="module")
+def imgpass_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """The one-source temp tree (see the module docstring): the app
+    server runs on the same host, so the paths are visible to it.
+    The directory NAME is the source name (``kind=local``, phase 38).
+    EXACTLY ONE file: the PyMuPDF-built ``imgpass.pdf`` (10 text
+    pages + 4 distinct non-blank images — see :func:`_build_pdf`)."""
+    root = tmp_path_factory.mktemp("bor_imgprog")
+    src = root / SOURCE
+    src.mkdir()
+    _build_pdf(src / PDF_NAME)
+    assert len(list(src.rglob("*"))) == N_FILES
+    return src
+
+
+def _app_env(base_url: str, tmp_path_factory: pytest.TempPathFactory) -> dict[str, str]:
+    """The per-module app env (the conftest pattern, cf.
+    ``test_oneshot_llm_retry.py`` / ``test_kb_tree.py``): the
+    ``base_url`` picks the LLM (the slow proxy in front of the mock),
+    ``BOR_GIT_SOURCES`` forced empty (the sync source is this suite's
+    own DB-registered local row), the sizing knobs pinned to the code
+    defaults (the leak guards — see the module constants), the
+    phase-126 knobs pinned EXPLICITLY (``BOR_IMPORT_EXTENSIONS`` with
+    the ``pdf`` token — without it the walk is blind to the fixture —
+    and ``BOR_IMAGES=true`` — the embedded pass is the span under
+    test), the E2E ``BOR_PDF_PICTORIAL`` off-pin (the house
+    convention — the suite's span is the embedded pass; the fixture's
+    ≥100-char pages keep the pictorial pass a no-op even if the knob
+    leaked on), and the other leak-guarded code defaults."""
+    env = dict(os.environ)
+    env.pop("DEBUGPY", None)
+    env["BOR_ENVIRONMENT"] = "e2e"
+    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
+    env["BOR_LLM_BASE_URL"] = (
+        "https://aipi.reeseapps.com/v1" if USE_REAL_LLM else base_url
+    )
+    # Mock-calibrated threshold (conftest pattern): this suite never
+    # asks the chat model anything — the gate is never on a path.
+    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
+    # Phase 67: instant retry waits + the code-default budget (the
+    # conftest leak-guard pattern).
+    env["BOR_LLM_RETRY_DELAY"] = "0"
+    env["BOR_LLM_RETRIES"] = str(_Settings.model_fields["llm_retries"].default)
+    env.setdefault(
+        "BOR_DATABASE_URL",
+        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
+    )
+    # Phase 16: admin auth must be set or create_app() refuses to boot.
+    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
+    env["BOR_SESSION_SECRET"] = SESSION_SECRET
+    # The repo's .env file carries the owner's BOR_GIT_SOURCES (the app
+    # reads it from cwd) — override it with an EMPTY value (the env var
+    # beats the .env file): the registry must hold EXACTLY the local
+    # directory this suite registers.
+    env["BOR_GIT_SOURCES"] = ""
+    # Leak guards (conftest pattern): an operator's local (gitignored)
+    # .env cannot leak corpus-specific settings into the app under
+    # test — including the chunking / batching knobs this suite's
+    # ground truth is computed from (see the module constants).
+    env["BOR_DOCS_REPO"] = ""
+    env["BOR_SUGGESTIONS"] = json.dumps(
+        _Settings.model_fields["suggestions"].default
+    )
+    env["BOR_INPUT_PLACEHOLDER"] = _Settings.model_fields["input_placeholder"].default
+    env["BOR_FOOTER_TEXT"] = _Settings.model_fields["footer_text"].default
+    env["BOR_CHUNK_TARGET_CHARS"] = str(CHUNK_TARGET)
+    env["BOR_CHUNK_OVERLAP_CHARS"] = str(CHUNK_OVERLAP)
+    env["BOR_EMBED_BATCH_SIZE"] = str(EMBED_BATCH_SIZE)
+    # Phase 126/122: the import scope WITH the ``pdf`` token + the
+    # embedded-image toggle ON (the span under test) + the E2E
+    # pictorial off-pin + the suite-private image home (the app under
+    # test must not write copies into the owner's real ~/bor-sources).
+    env["BOR_IMPORT_EXTENSIONS"] = "md,markdown,txt,pdf"
+    env["BOR_IMAGES"] = "true"
+    env["BOR_PDF_PICTORIAL"] = "0"
+    env["BOR_IMAGE_DIR"] = str(tmp_path_factory.mktemp("bor_imgprog_images"))
+    env["BOR_SOURCES_DIR"] = str(tmp_path_factory.mktemp("bor_checkouts"))
+    return env
+
+
+@pytest.fixture(scope="module")
+def slow_llm(mock_llm: int) -> Iterator[int]:
+    """The delay-injecting reverse proxy in front of the mock LLM
+    (``tests/e2e/slow_llm.py``) — the timing fixture for the whole
+    suite: it delays EVERY LLM call, vision describes included, so
+    each in-flight image pipeline's position outlives the UI's 2 s
+    poll (see ``SLOW_DELAY_S`` + the module docstring's timing
+    fixture paragraph)."""
+    env = dict(os.environ)
+    env.pop("DEBUGPY", None)
+    env["SLOW_LLM_DELAY_S"] = SLOW_DELAY_S
+    env["E2E_MOCK_PORT"] = str(MOCK_PORT)
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "tests.e2e.slow_llm:app",
+         "--host", "127.0.0.1", "--port", str(SLOW_PORT), "--log-level", "warning"],
+        cwd=REPO,
+        env=env,
+    )
+    try:
+        _wait_http(f"{SLOW_URL}/v1/models")
+        yield SLOW_PORT
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+
+
+@pytest.fixture(scope="module")
+def app_server(
+    mock_llm: int,
+    slow_llm: int,
+    imgpass_dir: Path,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> Iterator[str]:
+    """The SLOW app under test (the whole suite):
+    ``BOR_LLM_BASE_URL`` is the slow proxy in front of the mock (the
+    timing fixture — see the module docstring). ``imgpass_dir`` is a
+    dependency only for the fixture ordering (the temp tree exists
+    before the first test). Skips the module (clear instructions)
+    when Postgres is down — the sync needs the KB tables."""
+    env = _app_env(f"{SLOW_URL}/v1", tmp_path_factory)
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "app.main:app",
+         "--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning"],
+        cwd=REPO,
+        env=env,
+    )
+    try:
+        _wait_http(f"{APP_URL}/api/health")
+        body = httpx.get(f"{APP_URL}/api/health", timeout=5).json()
+        if body["db"] != "up":
+            pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
+        yield APP_URL
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+
+
+@pytest.fixture(scope="module")
+def app_url(app_server: str) -> str:
+    return app_server
+
+
+def _truncate_all() -> None:
+    """Fresh registry + KB (the E2E isolation pattern): the E2E suites
+    share one Postgres, so a leftover git_sources row or document would
+    pollute the file counts + the counts line the syncs and the
+    assertions pin exactly. ``sources_meta`` (the KB generation counter)
+    resets with the KB — the sync re-creates the row."""
+    with SessionLocal() as db:
+        db.execute(
+            text(
+                "TRUNCATE chunks, documents, query_log, steering_notes, "
+                "kb_overview, git_sources, sources_meta, folder_summaries"
+            )
+        )
+        db.commit()  # without the commit the TRUNCATE rolls back (the house pattern)
+
+
+def _wait_sync_settled(base_url: str, timeout_s: float = 180.0) -> None:
+    """No background sync may leak across tests (the run state lives in
+    the app's memory, and a still-running run would keep importing into
+    the NEXT test's truncated KB): wait for the app's sync status to
+    be non-running BEFORE the truncate. Own admin session (the
+    endpoint is admin-only) — usually a no-op: the tests settle only
+    after their run's terminal state."""
+    with httpx.Client(base_url=base_url, timeout=5.0) as client:
+        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
+        assert r.status_code == 204, r.text
+        deadline = time.monotonic() + timeout_s
+        while time.monotonic() < deadline:
+            r = client.get("/api/sync/status")
+            assert r.status_code == 200, r.text
+            if r.json()["state"] != "running":
+                return
+            time.sleep(0.2)
+    raise AssertionError(f"a sync was still running at a test boundary: {base_url}")
+
+
+@pytest.fixture(autouse=True)
+def _clean(app_url: str) -> Iterator[None]:
+    """Per-test KB isolation (the phase-96 cleanup pattern): the
+    shared E2E Postgres is truncated before AND after every test —
+    each test seeds its own source row and runs its own sync, so the
+    KB rows + folder_summaries rows of the temp source are exactly
+    this test's own. The pre-wait keeps a leaked run from the previous
+    test (the module app is shared) from racing the truncate."""
+    _wait_sync_settled(app_url)
+    _truncate_all()
+    yield
+    _truncate_all()
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _register_source(base_url: str, path: str) -> None:
+    """Register the temp directory through the authenticated API (the
+    ``test_local_directory_sources.py`` / phase-94 pattern — a plain
+    directory is an API/DB-only operation)."""
+    with httpx.Client(base_url=base_url, timeout=30.0) as client:
+        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
+        assert r.status_code == 204, r.text
+        r = client.post(
+            "/api/git-sources", json={"kind": "local", "path": path}
+        )
+        assert r.status_code == 201, r.text
+
+
+def _wait_terminal(app_url: str, timeout_s: float = 180.0) -> dict[str, Any]:
+    """Poll the status endpoint (own admin session) until the
+    UI-/API-triggered run reaches a terminal state — the source of
+    truth for the settle (the UI's 2 s poll renders it at most one
+    tick later). A FAILED terminal fails the test with the terminal
+    body (the ``detail`` / error text) — never an opaque label
+    timeout."""
+    with httpx.Client(base_url=app_url, timeout=30.0) as client:
+        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
+        assert r.status_code == 204, r.text
+        deadline = time.monotonic() + timeout_s
+        body: dict[str, Any] = {}
+        while time.monotonic() < deadline:
+            r = client.get("/api/sync/status")
+            assert r.status_code == 200, r.text
+            body = r.json()
+            if body["state"] in ("success", "failed"):
+                assert body["state"] == "success", body
+                return body
+            time.sleep(0.2)
+    raise AssertionError(f"the run never reached a terminal state: {body}")
+
+
+def _start_sync_click(page: Page, app_url: str, sync_dir: Path) -> None:
+    """Register the source, land the admin on the sources view, wait
+    for the settled boot label (the phase-64 pattern — also proves the
+    phase-32 boot re-attach completed, so no later re-render can
+    clobber the running state), then click **Sync sources** and pin
+    the click's immediate 202 state (the bare "Syncing…" — the count
+    AND the chunk parts hidden: no file yet, no batch in flight — and
+    the image part is absent too: no PDF's vision span has started,
+    ``images_total == 0``)."""
+    _register_source(app_url, str(sync_dir))
+    login(page, app_url)  # lands on /sources.html (the button's home)
+
+    btn = page.locator("#sync-btn")
+    expect(btn).to_be_visible(timeout=30_000)
+    expect(page.locator("#sync-label")).to_have_text(IDLE_OR_SETTLED)
+    expect(btn).to_be_enabled()
+
+    btn.click()
+
+    # The click's immediate state (the 202 moment — the bare prefix,
+    # the phase-64 contract): disabled, aria-busy, spinning icon, and
+    # the label's MAIN part alone ("Syncing…" — the 202 click carries
+    # no status, so the pinned parts hide, phase 128/129 writers).
+    # Phase 149 (task 01): the running detail rides the box — the 202's
+    # bare label is in #sync-progress (the button is the spinner).
+    expect(btn).to_be_disabled()
+    expect(btn).to_have_attribute("aria-busy", "true")
+    expect(btn.locator(".sync-icon")).to_have_class(re.compile(r"\bis-spinning\b"))
+    expect(page.locator("#sync-progress")).to_have_text("Syncing…")
+    expect(page.locator(".sync-label-count")).to_be_hidden()
+    expect(page.locator(".sync-label-chunk")).to_be_hidden()
+    expect(page.locator("#sync-error-banner")).to_be_hidden()
+
+
+# ---------------------------------------------------------------------------
+# 1. The anti-freeze proof — the label advances through the PDF's
+#    image span (≥3 distinct positions, monotonic, chunk part earlier,
+#    the A4 title/aria, the settled clearing)
+# ---------------------------------------------------------------------------
+
+
+def test_label_advances_through_the_image_span(
+    page: Page, app_url: str, imgpass_dir: Path
+) -> None:
+    """A fresh admin page on ``/sources.html`` clicks **Sync sources**;
+    the test then densely samples (≈ every 300 ms) the rendered
+    ``#sync-progress-text`` — NOT the API (the rendered DOM is the
+    observable; the UI's own 2 s poll is the writer) — collecting the
+    DISTINCT values in first-seen order until the settled counts line,
+    and asserts the phase-152 contract end to end (see the module
+    docstring's test mapping): the image part climbs (≥3 distinct
+    positions, starting at 1, non-decreasing, never above the total,
+    reaching it), the chunk part appeared earlier (and persists
+    through the image span with the file's TRUE chunk total), the
+    A4 title/aria ride the full string on every running sample, and
+    the settle clears the image position (the counts line, the
+    terminal label, the cleared endpoint position — no stale
+    position in a terminal state)."""
+    page.set_default_timeout(30_000)
+
+    # The ground truth — computed from the PDF ON DISK with the SAME
+    # parser + function + arguments the importer uses (see the helper
+    # docstrings): the TRUE chunk count + the batch count (the task's
+    # sizing bar — ≥2 embed batches, so the chunk part's climb is a
+    # real sequence).
+    pdf_path = imgpass_dir / PDF_NAME
+    chunks = _expected_chunks(pdf_path)
+    n_chunks = len(chunks)
+    assert n_chunks >= EMBED_BATCH_SIZE + 1, n_chunks  # 2+ embed batches
+    assert _expected_batch_count(chunks) >= 2  # the sizing bar
+    # The happy path's premise: a clean text layer (the phase-133
+    # fallback never fires) — the embedded-image pass is the file's
+    # one and only vision span (the ≥100-char pages keep the phase-140
+    # gate candidate-free even under the app's pinned-off knob).
+    assert is_usable_pdf_text(extract_pdf_text(pdf_path))
+
+    _start_sync_click(page, app_url, imgpass_dir)
+
+    # The observation loop (the task's "poll the page, not the API"):
+    # dense samples of the rendered box text (≈ 300 ms — the evaluate
+    # round-trip + the sleep) in first-seen order, each sample read
+    # atomically with the button's title/aria (A4 checked per running
+    # sample), until the settled counts line lands (or the generous
+    # deadline — a failed run then fails LOUDLY in _wait_terminal
+    # with the terminal body, never an opaque timeout).
+    seen: list[str] = []
+    last: str | None = None
+    deadline = time.monotonic() + 150
+    while time.monotonic() < deadline:
+        snap = page.evaluate(_JS_SAMPLE)
+        assert snap is not None
+        if snap["running"]:
+            # A4 — while a run is in flight, the title + the
+            # aria-label carry the SAME untruncated full string as the
+            # box (the "· image" part included).
+            assert snap["title"] == snap["box"], snap
+            assert snap["aria"] == snap["box"], snap
+        if snap["box"] != last:
+            seen.append(snap["box"])
+            last = snap["box"]
+        if snap["box"] == SETTLE_TEXT:
+            break
+        time.sleep(0.2)
+
+    # --- the settle (the UI's 2 s poll renders the terminal at most one
+    # tick after it lands; the endpoint is the source of truth) ---
+    assert last == SETTLE_TEXT, f"the run never settled to the counts line: {seen[-5:]}"
+    terminal = _wait_terminal(app_url)
+    # The counts line's ground truth: the PDF + its 4 embedded-image
+    # docs — one file, 5 added.
+    assert terminal["detail"]["added"] == 1 + N_IMAGES, terminal
+    # The keep-final-file-counts split (the phase-64/129 convention,
+    # restated for the phase-152 sub-position): the FILE counts keep
+    # their final values, the sub-file chunk + image positions clear
+    # with the file (0/0 in the terminal — no stale position), the
+    # phase clears.
+    assert terminal["files_done"] == N_FILES and terminal["files_total"] == N_FILES, terminal
+    assert terminal["current_file"] is None, terminal
+    assert terminal["current_chunk"] == 0 and terminal["chunks_total"] == 0, terminal
+    assert terminal["images_done"] == 0 and terminal["images_total"] == 0, terminal
+    assert terminal["phase"] is None, terminal
+
+    # --- the run was real (the prelude + the import span were
+    # sampled) ---
+    assert any(t == "Syncing…" for t in seen), f"no bare prelude label: {seen[:6]}"
+    import_labels = [t for t in seen if t.startswith("Syncing… ")]
+    assert any(f"({N_FILES}/{N_FILES})" in t for t in import_labels), (
+        f"no file count part: {import_labels[:6]}"
+    )
+
+    # --- the chunk part appeared EARLIER in the run (the two
+    # positions are distinct) ---
+    chunk_idx = next(
+        (i for i, t in enumerate(seen) if CHUNK_PART_RE.search(t)), None
+    )
+    assert chunk_idx is not None, f"no chunk part observed: {seen}"
+    first_chunk = CHUNK_PART_RE.search(seen[chunk_idx])
+    assert first_chunk is not None, seen[chunk_idx]
+    # The chunk part's total is the file's TRUE chunk count (not a
+    # batch count — the phase-129 A6 assumption, the E2E lens).
+    assert int(first_chunk.group(2)) == n_chunks, (first_chunk.group(0), n_chunks)
+
+    # --- the image part: ≥3 distinct positions (the anti-freeze
+    # proof — the owner's freeze is a single repeated label), the
+    # positions START at 1, are NON-DECREASING in observation order,
+    # never EXCEED the total (the post-dedupe/post-gate count — the
+    # work actually about to happen), and REACH it ---
+    def _image_part(t: str) -> re.Match[str]:
+        m = IMAGE_PART_RE.search(t)
+        assert m is not None, t  # the list comprehension's filter guarantees it
+        return m
+
+    image_items = [
+        (i, _image_part(t))
+        for i, t in enumerate(seen)
+        if IMAGE_PART_RE.search(t)
+    ]
+    assert len(image_items) >= 3, (
+        f"fewer than 3 distinct image positions observed: "
+        f"{[m.group(0) for _, m in image_items]}"
+    )
+    first_image_idx = image_items[0][0]
+    assert chunk_idx < first_image_idx, (chunk_idx, first_image_idx)
+    positions = [(int(m.group(1)), int(m.group(2))) for _, m in image_items]
+    assert positions[0][0] == 1, f"the climb does not start at 1: {positions}"
+    assert all(m == N_IMAGES for _, m in positions), positions
+    assert all(n <= N_IMAGES for n, _ in positions), positions
+    ns = [n for n, _ in positions]
+    assert ns == sorted(ns), f"the climb is backwards somewhere: {positions}"
+    assert N_IMAGES in ns, f"the climb never reached the total: {positions}"
+    # The two positions COEXIST (distinct and both may show — "… (1/1)
+    # · chunk 19/20 · image 3/4"): every image-span label carries the
+    # chunk part with it (the last chunk position persists through the
+    # file's image span — the per-file reset + the phase transitions
+    # clear it, not the end of the chunk span).
+    for i, _ in image_items:
+        cm = CHUNK_PART_RE.search(seen[i])
+        assert cm is not None, f"an image-span label lost the chunk part: {seen[i]}"
+        assert int(cm.group(2)) == n_chunks, (cm.group(0), n_chunks)
+
+    # --- the settle: the counts line, the terminal label, the parts
+    # cleared + empty, and the IMAGE part GONE (no stale position in a
+    # terminal state — the endpoint pin above is the server-side
+    # half) ---
+    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=30_000)
+    btn = page.locator("#sync-btn")
+    expect(btn).to_be_enabled()
+    expect(btn).not_to_have_attribute("aria-busy")
+    expect(page.locator(".sync-label-count")).to_be_hidden()
+    expect(page.locator(".sync-label-count")).to_have_text("")
+    expect(page.locator(".sync-label-chunk")).to_be_hidden()
+    expect(page.locator(".sync-label-chunk")).to_have_text("")
+    # The counts line rides the box (phase 150 — the sole sync text
+    # surface): the image part is not in it, not in the title, not in
+    # the aria (the settle restores the ship aria value).
+    box = page.locator("#sync-progress-text").inner_text()
+    assert box == SETTLE_TEXT, box
+    assert "· image" not in box and "· image" not in (btn.get_attribute("title") or "")
+    assert btn.get_attribute("aria-label") == "Sync sources", btn.get_attribute("aria-label")
+    expect(page.locator("#sync-error-banner")).to_be_hidden()
+
+
+# ---------------------------------------------------------------------------
+# 2. The never-stale contract — a mid-run page reload re-attaches with
+#    the current image position visible
+# ---------------------------------------------------------------------------
+
+
+def test_mid_run_reload_re_attaches_with_image_position(
+    page: Page, app_url: str, imgpass_dir: Path
+) -> None:
+    """The admin clicks **Sync sources** (the run is started the way
+    the owner starts it — the click is what starts the page's 2 s
+    poll; the reload is the contract under test), the page's own 2 s
+    poll renders the image span (the FIRST observed position is
+    ``1/4`` — the whole span is ahead, so the reload + the load-time
+    re-attach happen with the span in flight), the test reloads the
+    page mid-run, and the load-time re-attach (``initSyncButton`` —
+    a running sync re-enters with the WHOLE status, phase 98's
+    never-stale contract) renders the CURRENT ``· image n/4`` with
+    the A4 title/aria; the run then settles to the counts line (the
+    image part gone — no stale position)."""
+    page.set_default_timeout(30_000)
+    _start_sync_click(page, app_url, imgpass_dir)
+
+    # The image span, rendered by the page's own 2 s poll (the span
+    # starts after the prelude + the file's embed batches — ≈ 21 s
+    # under the slow proxy; each position holds 4.5 s — the describe
+    # + the image doc's 2 embed batches).
+    snap = page.wait_for_function(
+        _JS_IMAGE_SNAP, arg=N_IMAGES, timeout=60_000
+    ).json_value()
+    # The FIRST observed position is 1 — the whole span is ahead (the
+    # reload's re-attach is guaranteed to catch the span in flight:
+    # ≈ 18 s remain, the local reload + the load-time status fetch
+    # land well inside it).
+    assert snap["posN"] == 1, snap
+    # A4 — the pre-reload running state is untruncated on the title +
+    # the aria (the "· image" part included).
+    assert snap["title"] == snap["box"] and "· image" in snap["title"], snap
+    assert snap["aria"] == snap["box"], snap
+
+    # The mid-run reload (the session cookie survives — the re-attach
+    # is the admin's, phase 98).
+    page.reload()
+
+    # The load-time re-attach: a running sync re-enters with the WHOLE
+    # status — the CURRENT image position is visible (the never-stale
+    # contract), the A4 title/aria ride it.
+    snap2 = page.wait_for_function(
+        _JS_IMAGE_SNAP, arg=N_IMAGES, timeout=30_000
+    ).json_value()
+    assert snap2["title"] == snap2["box"] and "· image" in snap2["title"], snap2
+    assert snap2["aria"] == snap2["box"], snap2
+
+    # Then the settle: the counts line (the image part gone — the
+    # terminal clears the position, test 1 pins the endpoint half),
+    # the terminal label, the parts cleared + empty.
+    expect(page.locator("#sync-progress-text")).to_have_text(
+        SETTLE_TEXT, timeout=SETTLE_TIMEOUT_MS
+    )
+    expect(page.locator("#sync-label")).to_have_text(SYNCED_LABEL, timeout=30_000)
+    btn = page.locator("#sync-btn")
+    expect(btn).to_be_enabled()
+    expect(btn).not_to_have_attribute("aria-busy")
+    expect(page.locator(".sync-label-count")).to_be_hidden()
+    expect(page.locator(".sync-label-chunk")).to_be_hidden()
+    assert "· image" not in (btn.get_attribute("title") or "")
+    expect(page.locator("#sync-error-banner")).to_be_hidden()
+
+    terminal = _wait_terminal(app_url)
+    assert terminal["detail"]["added"] == 1 + N_IMAGES, terminal
+    assert terminal["images_done"] == 0 and terminal["images_total"] == 0, terminal

tests/integration/test_sync_api.py

@@ -93,6 +93,7 @@ import asyncio
 import logging
 import os
 import subprocess
+import textwrap
 import threading
 import time
 from collections.abc import Callable, Iterator, Sequence
@@ -100,6 +101,7 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
+import pymupdf as fitz
 import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy import select, text
@@ -309,6 +311,12 @@ class FakeImportSources:
         # regresses) — the canned fake never fires it (the runner's
         # status chunk fields stay 0/0, like a no-batch import).
         self.chunk_hooks: list[object] = []
+        # Phase 152 (task 03): the image-position hook the runner
+        # passes (a live closure while wired, None if the wiring
+        # regresses) — the canned fake never fires it (the status
+        # image fields stay 0/0, like the chunk fields of a no-batch
+        # import).
+        self.image_hooks: list[object] = []
 
     async def __call__(
         self,
@@ -320,6 +328,7 @@ class FakeImportSources:
         session: Session | None = None,
         progress: Callable[[str, str, int, int], None] | None = None,
         on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
+        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
         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,  # phase 106
@@ -329,6 +338,7 @@ class FakeImportSources:
         self.prune_flags.append(prune)
         self.progress_hooks.append(progress)
         self.chunk_hooks.append(on_chunk_progress)
+        self.image_hooks.append(on_image_progress)
         self.ignore_maps.append(ignore_by_root or {})
         self.include_hidden_maps.append(include_hidden_by_root or {})
         self.doc_dates_maps.append(doc_dates_by_root or {})
@@ -500,6 +510,10 @@ def test_admin_sync_success_reports_full_detail(
         # (the dataclass defaults).
         "current_chunk": 0,
         "chunks_total": 0,
+        # Phase 152 (task 01): the image-position keys — 0/0 idle
+        # (the dataclass defaults).
+        "images_done": 0,
+        "images_total": 0,
         # Phase 98 (task 01): the phase-machine keys — null/0/0/0 idle.
         "phase": None,
         "current_summary": None,
@@ -656,13 +670,23 @@ def test_double_trigger_while_running_returns_409(
 class _GatedImport:
     """An import that fires the runner's progress hook once, then
     parks on a threading gate — the test reads the status mid-import
-    (the phase ``"import"`` + the file hook)."""
+    (the phase ``"import"`` + the file hook).
+
+    Phase 152 (task 03): optionally ALSO fires the in-flight file's
+    image-position pairs (``image_pairs``) through the runner's OWN
+    ``on_image_progress`` closure: pair one lands before the main park
+    (readable mid-run), every later pair after the main park + its own
+    park (``image_release``), so each position is readable on the
+    status between polls (the phase-129 chunk idiom, mirrored for the
+    vision-image position)."""
 
     def __init__(
         self,
         summary: ImportSummary,
         started: threading.Event,
         release: threading.Event,
+        image_pairs: Sequence[tuple[int, int]] = (),
+        image_release: threading.Event | None = None,
     ) -> None:
         self.summary = summary
         self.started = started
@@ -672,6 +696,13 @@ class _GatedImport:
         # passes (recorded for the wiring pin — the gated fake never
         # fires it, so the status chunk fields stay 0/0).
         self.chunk_hook: Callable[[int, int], None] | None = None
+        # Phase 152 (task 03): the image-position hook the runner
+        # passes (recorded for the wiring pin) + the optional image
+        # pairs (see the class docstring).
+        self.image_hook: Callable[[int, int], None] | None = None
+        self.image_pairs = list(image_pairs)
+        self.image_release = image_release
+        self.image_hook_calls: list[tuple[int, int]] = []
 
     async def __call__(
         self,
@@ -683,6 +714,7 @@ class _GatedImport:
         session: Session | None = None,
         progress: Callable[[str, str, int, int], None] | None = None,
         on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
+        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
         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,  # phase 106
@@ -691,6 +723,25 @@ class _GatedImport:
             progress("repo", "notes/deep.md", 1, 3)
             self.hook_calls.append(("repo", "notes/deep.md", 1, 3))
         self.chunk_hook = on_chunk_progress
+        self.image_hook = on_image_progress
+        if on_image_progress is not None and self.image_pairs:
+            # Phase 152 (task 03): the in-flight PDF's image-position
+            # pairs, fired through the runner's OWN hook closure (the
+            # closure under test) — pair one before the main park, each
+            # later pair after the main park + its own park (each park
+            # lets the test read the status between polls). A wiring
+            # regression to None skips this branch and the pair test
+            # times out.
+            on_image_progress(*self.image_pairs[0])
+            self.image_hook_calls.append(self.image_pairs[0])
+            self.started.set()
+            await asyncio.to_thread(self.release.wait)
+            for pair in self.image_pairs[1:]:
+                on_image_progress(*pair)
+                self.image_hook_calls.append(pair)
+                if self.image_release is not None:
+                    await asyncio.to_thread(self.image_release.wait)
+            return self.summary
         self.started.set()
         await asyncio.to_thread(self.release.wait)
         return self.summary
@@ -867,6 +918,305 @@ def test_status_reports_the_phase_machine_across_the_run(
     assert gated_import.hook_calls == [("repo", "notes/deep.md", 1, 3)]
 
 
+def test_image_position_climbs_between_polls_mid_run(
+    sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
+) -> None:
+    """Phase 152 (task 03): while a PDF's vision span is in flight,
+    the status carries the in-flight image position — the pairs are
+    fired through the runner's OWN ``_image_hook`` closure (task 03's
+    wiring: ``import_sources(..., on_image_progress=_image_hook)``),
+    and the position CLIMBS between two polls of
+    ``GET /api/sync/status`` (1/15 → 7/15 — monotonic, never above
+    the total, no backwards jumps — the owner's stuck label now
+    moves), paired with the file the ``progress`` hook named (the
+    hooks are paired — the position belongs to that file); the
+    success terminal clears the position to 0/0 while the FILE
+    counts keep their final values (the convention split — same as
+    the chunk position)."""
+    repo_url = f"file://{tmp_path / 'repo.git'}"
+    _seed(db, repo_url)
+    _stub_env(monkeypatch)
+    monkeypatch.setattr(
+        sync_api,
+        "get_settings",
+        lambda: _settings(sources_dir=str(tmp_path / "bor")),
+    )
+    _, fake_clone = _fake_clone()
+    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
+    _stub_probe(monkeypatch)
+    image_release = threading.Event()
+    gated_import = _GatedImport(
+        ImportSummary(files=3, added=1, updated=1, unchanged=1),
+        threading.Event(),
+        threading.Event(),
+        image_pairs=[(1, 15), (7, 15)],  # two describe calls of a 15-image span
+        image_release=image_release,
+    )
+    monkeypatch.setattr(sync_api, "import_sources", gated_import)
+    monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
+    monkeypatch.setattr(sync_api, "generate_folder_summaries", FakeFolderSummaries())
+
+    _login(sync_client)
+    # Idle: the image keys ride along as 0/0 (the dataclass defaults).
+    body = sync_client.get("/api/sync/status").json()
+    assert body["state"] == "idle"
+    assert body["images_done"] == 0 and body["images_total"] == 0
+
+    assert sync_client.post("/api/sync").status_code == 202
+    try:
+        assert gated_import.started.wait(5.0), "the run never reached the first image pair"
+        # Poll 1 — the import phase, pair one live: the position is
+        # 1/15, paired with the file the progress hook named; the
+        # chunk position is 0/0 (this canned file has no embed
+        # batches — the two positions are independent).
+        s = sync_client.get("/api/sync/status").json()
+        assert s["state"] == "running"
+        assert s["phase"] == "import"
+        assert s["current_file"] == "repo/notes/deep.md"
+        assert s["files_done"] == 1 and s["files_total"] == 3
+        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+        assert s["images_done"] == 1 and s["images_total"] == 15
+        # Release pair one: the import fires pair two (7/15 — the
+        # NEXT describe call of the same 15-image span) and parks
+        # again.
+        gated_import.release.set()
+        deadline = time.monotonic() + 5.0
+        while time.monotonic() < deadline:
+            s = sync_client.get("/api/sync/status").json()
+            if s["images_done"] == 7:
+                break
+            assert s["state"] == "running", (
+                f"unexpected state {s['state']!r} while waiting for pair two: {s}"
+            )
+            time.sleep(0.05)
+        else:
+            pytest.fail("the second image pair never landed on the status")
+        # Poll 2 — the position CLIMBED (1 → 7, monotonic — no
+        # backwards jump, never above the total): the label is
+        # moving, not frozen.
+        assert s["state"] == "running"
+        assert s["phase"] == "import"
+        assert s["current_file"] == "repo/notes/deep.md"
+        assert s["images_done"] == 7 and s["images_total"] == 15
+        assert 0 < s["images_done"] <= s["images_total"]
+        image_release.set()
+    finally:
+        gated_import.release.set()
+        image_release.set()
+
+    body = _poll(sync_client, "success")
+    # Terminal: the image position clears (0/0) ... the FILE counts
+    # KEEP their final values (the convention split — the position is
+    # a sub-position of a done file, like the chunk position).
+    assert body["images_done"] == 0 and body["images_total"] == 0
+    assert body["current_file"] is None
+    assert body["files_done"] == 1 and body["files_total"] == 3
+    # Wiring: BOTH pairs went through the runner's OWN live closure
+    # (the status reads above came through it — a wiring regression
+    # to None would have skipped the firing branch and timed out),
+    # and the file hook fired once.
+    assert gated_import.image_hook is not None
+    assert gated_import.image_hook_calls == [(1, 15), (7, 15)]
+    assert gated_import.hook_calls == [("repo", "notes/deep.md", 1, 3)]
+
+
+# --- phase 152 (task 04): the full-run image-position monotonicity -------
+
+
+def _kb_solid_png(value: int) -> bytes:
+    """A 40x30 solid-color PNG (channel *value*): every value used
+    below stays under the phase-151 blank gate's 235 mean floor
+    (kept, never blank-skipped), and distinct values are distinct
+    images (the parser's dedupe keeps every one)."""
+    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 40, 30), 0)
+    pix.clear_with(value)
+    return pix.tobytes("png")
+
+
+def _kb_image_pdf(path: Path, values: Sequence[int]) -> None:
+    """A REAL one-page PDF (the fixture KB's image-bearing file): a
+    text block comfortably over the phase-140 100-char sparseness
+    bar (a USABLE layer — the phase-133 gate stays closed, zero
+    pictorial candidates even with the toggle on — the "no other
+    vision span" class) carrying *values* the solid-color embedded
+    images (one placement each, distinct)."""
+    doc = fitz.open()
+    page = doc.new_page()
+    text = (
+        "The quadlet unit file defines the container interface, the "
+        "image, the network, and the restart policy. Every key maps "
+        "to a docker flag in the generated service file, and comments "
+        "use the usual semicolon form throughout the whole document."
+    )
+    for i, line in enumerate(textwrap.wrap(text, 60)):
+        page.insert_text((72, 72 + i * 14), line)
+    y = 300
+    for value in values:
+        page.insert_image(fitz.Rect(72, y, 112, y + 30), stream=_kb_solid_png(value))
+        y += 40
+    doc.save(path)
+    doc.close()
+
+
+class _GatedVisionLLM(FakeEmbedder):
+    """The phase-152 (task 04) mock LLM — the deterministic
+    in-process ``FakeEmbedder`` whose VISION calls (the multimodal
+    describe shape ``describe_image`` sends) PARK on a threading
+    gate: the run parks inside the PDF's embedded-image span, so the
+    test can read the status between the describe calls
+    (deterministic — no sleeps, no wall-clock dependence; the gate
+    holds the run until the test releases it). The hook fired BEFORE
+    the parked call (the task-02 contract), so the position is live
+    while the run is parked. Text (``lite``) calls keep the
+    ``FakeEmbedder`` behaviour; the embeds are the deterministic
+    vectors."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.vision_parks = 0  # the vision calls that parked (and released)
+        self.vision_release = threading.Event()
+
+    async def chat(
+        self, messages: list[dict[str, Any]], model: str | None = None
+    ) -> str:
+        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
+        if isinstance(user, list):
+            # One slow vision describe call — the gate (clear before
+            # the wait — a latching set would otherwise leak across
+            # calls): the test releases after reading the position.
+            self.vision_release.clear()
+            self.vision_parks += 1
+            await asyncio.to_thread(self.vision_release.wait)
+            return "A solid color swatch described."  # deterministic, non-empty
+        return await super().chat(messages, model)
+
+
+def test_image_position_monotonic_across_polls_over_a_real_pdf(
+    sync_client: TestClient,
+    monkeypatch: pytest.MonkeyPatch,
+    db: Session,
+    tmp_path: Path,
+    clean_documents: None,
+) -> None:
+    """Phase 152 (task 04): a FULL sync run over a fixture KB with
+    ONE image-bearing PDF (the REAL ``import_sources`` — the
+    runner's ``_image_hook`` is driven by the importer's own
+    embedded-image span; the LLM is the deterministic mock whose
+    vision calls park on a gate) — polling ``GET /api/sync/status``
+    from the test while the run is parked between the describe
+    calls, the observed ``images_done`` sequence is MONOTONICALLY
+    NON-DECREASING, never above ``images_total``, and ends at
+    ``images_total`` (no backwards jumps, no overshoot — the
+    owner's stuck label climbs one describe call at a time); the
+    position is paired with the file the ``progress`` hook named
+    (the hooks are paired), the success terminal clears it to 0/0
+    while the FILE counts keep their final values, and the KB
+    carries the PDF + its N image docs (the span really ran)."""
+    kb = tmp_path / "PictKB"
+    kb.mkdir()
+    (kb / "notes.md").write_text("# Notes\nthe local fixture\n", encoding="utf-8")
+    _kb_image_pdf(kb / "report.pdf", (90, 130, 170))  # 3 distinct images
+    n = 3
+    _seed_local(db, kb)
+    _stub_env(monkeypatch)
+    monkeypatch.setattr(
+        sync_api,
+        "get_settings",
+        lambda: _settings(sources_dir=str(tmp_path / "bor")),
+    )
+    clone_calls, fake_clone = _fake_clone()
+    monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
+    llm = _GatedVisionLLM()
+    llm.settings = Settings(
+        _env_file=None,  # pyright: ignore[reportCallIssue]
+        import_extensions="md,txt,pdf",
+        images=True,
+        image_dir=str(tmp_path / "images"),
+        pdf_page_dir=str(tmp_path / "pages"),
+    )
+    monkeypatch.setattr(sync_api, "LLMClient", lambda: llm)
+    _stub_probe(monkeypatch)  # the probe passes on the mock (no network)
+
+    _login(sync_client)
+
+    def status() -> dict:
+        return sync_client.get("/api/sync/status").json()
+
+    def wait_for(predicate, what: str, timeout: float = 10.0) -> None:
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if predicate():
+                return
+            time.sleep(0.02)
+        raise AssertionError(f"timed out waiting for {what}")
+
+    # Idle: the image keys ride along as 0/0 (the dataclass
+    # defaults).
+    body = status()
+    assert body["state"] == "idle"
+    assert body["images_done"] == 0 and body["images_total"] == 0
+
+    assert sync_client.post("/api/sync").status_code == 202
+    try:
+        # The run reaches the PDF's vision span: the first hook fired
+        # (total = the span's N) and the first describe call parked
+        # the run (the gate).
+        wait_for(
+            lambda: status()["state"] == "running"
+            and status()["phase"] == "import"
+            and status()["images_total"] == n,
+            "the run to reach the PDF's vision span",
+        )
+        # Each describe call: the hook fired BEFORE the call, so the
+        # position is live while the run is parked in the call —
+        # observe it, release the call, let the next position climb.
+        observed: list[tuple[int, int]] = []
+        for expected in range(1, n + 1):
+            def position_ready(target: int = expected) -> bool:
+                # the default argument binds the loop variable (B023)
+                return status()["images_done"] == target
+
+            wait_for(position_ready, f"images_done to climb to {expected}")
+            s = status()
+            assert s["state"] == "running" and s["phase"] == "import"
+            # The position is paired with the file the progress hook
+            # named (the two hooks are paired — the position belongs
+            # to that file).
+            assert s["current_file"] == "PictKB/report.pdf"
+            observed.append((s["images_done"], s["images_total"]))
+            llm.vision_release.set()
+        # The run finishes the rest of the pipeline (the span is
+        # done — no more vision calls) and settles.
+        assert llm.vision_parks == n, "one park per describe call"
+    finally:
+        llm.vision_release.set()
+
+    body = _poll(sync_client, "success")
+    # The observed sequence: monotonically non-decreasing, never
+    # above the total, ending at the total (the gate made it exact —
+    # 1..N, no backwards jumps, no overshoot).
+    dones = [d for d, _ in observed]
+    assert dones == list(range(1, n + 1)), f"the sequence: {observed}"
+    assert all(d <= t for d, t in observed), "no overshoot"
+    assert dones[-1] == n, "ends at images_total"
+    assert {t for _, t in observed} == {n}, "one total for the whole span"
+    # Terminal: the image position clears (0/0) ... the FILE counts
+    # KEEP their final values (the convention split — the position
+    # is a sub-position of a done file, like the chunk position).
+    assert body["images_done"] == 0 and body["images_total"] == 0
+    assert body["current_file"] is None
+    assert body["files_done"] == 2 and body["files_total"] == 2
+    assert clone_calls == []  # nothing to clone — local is walked directly
+    # The KB carries the fixture md + the PDF + its N image docs —
+    # the span really ran (N new image documents, no failures).
+    assert body["detail"]["added"] == 2 + n
+    assert body["detail"]["errors"] == 0
+    docs = {(d["source"], d["path"])
+           for d in sync_client.get("/api/docs").json()["documents"]}
+    assert ("PictKB", "report.pdf") in docs
+    assert sum(1 for _, p in docs if "#page=" in p) == n
+
+
 def test_failed_terminal_after_the_summary_hook_clears_phase_keeps_counts(
     sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
 ) -> None:
@@ -1497,6 +1847,7 @@ def test_import_error_is_reported_with_credentials_masked(
         session: Session | None = None,
         progress: Callable[[str, str, int, int], None] | None = None,
         on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
+        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
         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,  # phase 106

tests/unit/test_frontend_sync_upload.py

@@ -127,6 +127,28 @@ setSyncProgress and pins the result mode behaviorally (a settle
 result line reveals the box, "" empties + hides it); the bare
 settle's row now shows the box LEFT ALONE (the settle label rides
 the button only — A2).
+
+Phase 152 (task 03) adds the label's FOURTH part — the vision-image
+position: fmtSyncPhaseLabel's import fall-through gains "· image
+n/m" (rendered while images_total > 0 — a PDF's vision span is in
+flight; n the 1-based image about to be described, m the file's
+total images, ONE continuous counter across the PDF's three vision
+spans), the overview/summaries branches return image: "" (the
+position belongs to the import phase only — the status's image
+fields are cleared at those transitions, so the branches never read
+them), and the bare builder stays image-free (the upload label and
+the 202/409 click stay bare). setSyncLabel(mainText, countText =
+"", chunkText = "", running = false, imageText = "") appends the
+image part LAST to the composed full string — so the box, the
+button title, and the running aria-label all carry
+"Syncing… <file> (2/3) · chunk 14/38 · image 3/15". The image part
+gets NO span of its own (the running button is the spinner — its
+parts are hidden, phase 149): it rides ONLY the composed string,
+while the chunk part keeps its .sync-label-chunk span; the two
+positions are distinct and each disappears independently. The node
+harness gains the two in-flight image rows (both parts at once —
+the owner's stuck label — and the image-only pictorial PDF) and
+pins the composed full string byte-for-byte.
 """
 from __future__ import annotations
 
@@ -298,8 +320,9 @@ def test_fmt_sync_label_returns_the_parts_contract() -> None:
 
 def test_set_sync_label_is_the_single_parts_writer() -> None:
     """setSyncLabel(mainText, countText = "", chunkText = "",
-    running = false) — Phase 128 (task 01) + Phase 129 (task 04):
-    the SINGLE point that writes #sync-label. The MAIN part gets
+    running = false, imageText = "") — Phase 128 (task 01) + Phase
+    129 (task 04) + Phase 152 (task 03): the SINGLE point that writes
+    #sync-label. The MAIN part gets
     mainText (textContent only — the house rule, no HTML strings);
     the COUNT part gets countText when present (textContent +
     hidden=false — the "(n/m)" position is visible, NEVER clipped)
@@ -309,11 +332,18 @@ def test_set_sync_label_is_the_single_parts_writer() -> None:
     hidden=true — the part renders only while a file's embed batch
     is in flight). Then the UNTRUNCATED full label — mainText +
     (countText ? " " + countText : "") + (chunkText ? " " + chunkText
-    : "") — rides the button title (A4 — the screen reader hears the
-    whole "Syncing… <file> (2/3) · chunk 14/38" there, and in the
-    box — see below). Phase 149 (task 01): the fourth parameter
+    : "") + (imageText ? " " + imageText : "") — rides the button
+    title (A4 — the screen reader hears the
+    whole "Syncing… <file> (2/3) · chunk 14/38 · image 3/15" there,
+    and in the box — see below). The image part (phase 152) gets NO
+    span of its own (the running button is the spinner — its parts
+    are hidden, phase 149): it rides only the composed full string.
+    Phase 149 (task 01): the fourth parameter
     `running` (the in-flight flag — the running entry passes true,
-    the settle the default false). Phase 150 (task 02): while
+    the settle the default false); phase 152 (task 03) adds the
+    fifth — `imageText` (the bare builders and the non-running calls
+    pass nothing, so it defaults "" and the part stays out of the
+    string). Phase 150 (task 02): while
     running, the SAME full label goes into the detail box
     (#sync-progress — the SOLE aria-live region, the announcer the
     screen reader hears) through the single box writer
@@ -324,10 +354,20 @@ def test_set_sync_label_is_the_single_parts_writer() -> None:
     running / the ship value "Sync sources" otherwise
     (overwrites/restores only — never removes)."""
     body = _fn(_js(), "setSyncLabel")
-    assert (
-        'function setSyncLabel(mainText, countText = "", chunkText = "", running = false)'
-        in body
-    ), "phase 149: the in-flight flag is the fourth parameter"
+    # the exact signature, byte-for-byte (concatenated — the literal
+    # is 105 columns, over the line cap): main, count, chunk, the
+    # phase-149 `running` flag (fourth), the phase-152 `imageText`
+    # (fifth — the bare builders + non-running calls pass nothing,
+    # so it defaults to the empty string).
+    expected_sig = (
+        'function setSyncLabel(mainText, countText = "", chunkText = "",'
+        ' running = false, imageText = "")'
+    )
+    assert expected_sig in body, (
+        "phase 149: the in-flight flag is the fourth parameter; phase"
+        " 152: the image part is the fifth, defaulting to the empty"
+        " string (the part stays out of the composed full string)"
+    )
     # the parts: textContent only (the house rule), null-guarded refs
     assert "syncLabelMain.textContent = mainText" in body
     assert "syncLabelCount.textContent = countText" in body
@@ -353,8 +393,12 @@ def test_set_sync_label_is_the_single_parts_writer() -> None:
         "const full =\n"
         "      mainText +\n"
         "      (countText ? ` ${countText}` : \"\") +\n"
-        "      (chunkText ? ` ${chunkText}` : \"\");"
-    ) in body, ("the full label is main + ' ' + count + ' ' + chunk (spaces = the CSS gaps)")
+        "      (chunkText ? ` ${chunkText}` : \"\") +\n"
+        "      (imageText ? ` ${imageText}` : \"\");"
+    ) in body, (
+        "the full label is main + ' ' + count + ' ' + chunk + ' ' +"
+        " image — the image part LAST (spaces = the CSS gaps)"
+    )
     i_full = body.find("const full")
     i_title = body.find("syncBtn.title = full")
     i_box = body.find("if (running) setSyncProgress(full);")
@@ -375,7 +419,8 @@ def test_set_sync_label_is_the_single_parts_writer() -> None:
 
 def test_fmt_sync_phase_label_is_the_exact_d2_parts_contract() -> None:
     """fmtSyncPhaseLabel(status) — the SYNC job's phase-aware label
-    (phase 98 task 02, D2; phase 128 task 01: the PARTS). "overview"
+    (phase 98 task 02, D2; phase 128 task 01: the PARTS; phase 152
+    task 03: the IMAGE part). "overview"
     → main exactly "Writing KB overview…" (no folder, no count part);
     "summaries" → main "Summarizing folders… <current_summary>" +
     count "(n/m)" — the folder part (one leading space +
@@ -385,7 +430,14 @@ def test_fmt_sync_phase_label_is_the_exact_d2_parts_contract() -> None:
     a total); anything else (the null prelude, "import") → the
     phase-64 fmtSyncLabel parts, byte-identical (same builder, same
     fields, same order). The UPLOAD job never enters here — its status
-    has no phase (D2)."""
+    has no phase (D2). Phase 152 (task 03): the import fall-through
+    is the ONLY branch that reads the status's image fields — "·
+    image n/m" renders while images_total > 0 (a PDF's vision span is
+    in flight — n the 1-based image about to be described, m the
+    file's total images) and hides at 0/0; the overview/summaries
+    branches return image: "" and never read the fields (the position
+    belongs to the import phase only — the runner clears it at those
+    transitions)."""
     body = _fn(_js(), "fmtSyncPhaseLabel")
     assert "function fmtSyncPhaseLabel(status)" in body
     # "overview": the exact copy, no folder, no counts.
@@ -421,14 +473,30 @@ def test_fmt_sync_phase_label_is_the_exact_d2_parts_contract() -> None:
     )
     assert "chunks_total" not in ov, "the overview branch never reads the chunk fields"
     assert "chunks_total" not in sum_, "the summaries branch never reads the chunk fields"
+    # phase 152 (task 03): the overview/summaries labels carry NO
+    # image part either (the position belongs to the import phase
+    # only) and never read the status's image fields (the runner
+    # clears them at those transitions).
+    assert 'image: ""' in ov, "the overview label carries no image part"
+    assert 'image: ""' in sum_, (
+        "the summaries label carries no image part (its (n/m) is its position)"
+    )
+    assert "images_total" not in ov, "the overview branch never reads the image fields"
+    assert "images_total" not in sum_, "the summaries branch never reads the image fields"
     # anything else (null prelude, "import"): byte-identical to today's
     # sync label — same builder, same fields, same order — plus the
     # phase-129 chunk part: "· chunk n/m" ONLY while chunks_total > 0
     # (a file's embed batch is in flight — n the 1-based first chunk
     # of the batch, m the file's total content chunks, batch
     # granularity), "" otherwise (0/0 before the first batch — the
-    # part hides). The fall-through is the ONLY branch that reads the
-    # status's chunk fields.
+    # part hides) + the phase-152 image part: "· image n/m" ONLY
+    # while images_total > 0 (a PDF's vision span is in flight — n
+    # the 1-based image about to be described, m the file's total
+    # images — ONE continuous counter across the PDF's three vision
+    # spans), "" otherwise (0/0 — the part hides). The fall-through
+    # is the ONLY branch that reads the status's chunk + image
+    # fields, and the two parts are independent (both may show,
+    # each disappears on its own 0/0).
     i_destr = body.find("const { main, count } = fmtSyncLabel(")
     assert -1 < i_destr < i_fall, "the fall-through destructures the phase-64 parts"
     fall = body[i_destr:]
@@ -444,11 +512,22 @@ def test_fmt_sync_phase_label_is_the_exact_d2_parts_contract() -> None:
     i_guard = fall.find("status.chunks_total > 0")
     i_else = fall.find(": \"\";", i_guard)
     assert -1 < i_guard < i_else, "0/0 (before the first batch) → the empty chunk part"
-    assert "return { main, count, chunk }" in fall
+    # phase 152 (task 03): the image part — "· image n/m" ONLY while
+    # images_total > 0, the exact template + the 0/0 → "" guard, and
+    # the return gains the fourth part.
+    assert "status.images_total > 0" in fall, (
+        "the image part renders only while a vision span is in flight"
+    )
+    assert "`· image ${status.images_done}/${status.images_total}`" in fall
+    i_iguard = fall.find("status.images_total > 0")
+    i_ielse = fall.find(": \"\";", i_iguard)
+    assert -1 < i_iguard < i_ielse, "0/0 (before the first image) → the empty image part"
+    assert "return { main, count, chunk, image }" in fall
     # the bare builder (the upload job + the 202/409 click) returns NO
-    # chunk part — those labels stay bare.
+    # chunk OR image part — those labels stay bare.
     bare = _fn(_js(), "fmtSyncLabel")
     assert "chunk" not in bare, "fmtSyncLabel stays chunk-free (upload + the click stay bare)"
+    assert "image" not in bare, "fmtSyncLabel stays image-free (upload + the click stay bare)"
 
 
 def test_enter_running_state_is_phase_aware_and_writes_the_parts() -> None:
@@ -460,13 +539,16 @@ def test_enter_running_state_is_phase_aware_and_writes_the_parts() -> None:
     endpoint has no phase) and the 202/409 click keep the bare
     fmtSyncLabel parts, byte-identical. Phase 128 (task 01): the entry
     itself no longer writes the label — it hands the parts to the
-    SINGLE writer (setSyncLabel(main, count, chunk) — phase 129:
-    the chunk part destructures with a "" default), which lands the
+    SINGLE writer (setSyncLabel(main, count, chunk, running, image) —
+    phase 129: the chunk part destructures with a "" default, phase
+    152: the image part does too), which lands the
     main part, the pinned count part (hidden when absent), the pinned
     chunk part (phase 129 — hidden when absent), and the FULL
     untruncated label on the button title + #sync-progress (A4 — set
-    for EVERY running label; CSS ellipsizes the main part only).
-    Phase 150 (task 02): the box write rides in the writer too
+    for EVERY running label; CSS ellipsizes the main part only —
+    phase 152: with the image part appended last when a PDF's vision
+    span is in flight; it gets no span of its own). Phase 150 (task
+    02): the box write rides in the writer too
     (setSyncLabel → setSyncProgress, running mode)."""
     body = _fn(_js(), "enterSyncRunningState")
     assert (
@@ -487,13 +569,16 @@ def test_enter_running_state_is_phase_aware_and_writes_the_parts() -> None:
     # the parts are handed to the single writer (A4 lives in it) —
     # phase 129: the chunk part destructures with a "" default (the
     # bare builders — the upload job, the 202/409 click — return no
-    # chunk part, so the part stays hidden there); phase 149: the
+    # chunk part, so the part stays hidden there); phase 152: the
+    # image part does too (it rides only the composed full string —
+    # no span); phase 149: the
     # call passes `running = true` (the detail box reveals + the
     # full-label aria-label ride in the writer).
-    assert "const { main, count, chunk = \"\" } =" in body, (
-        "the chunk part destructures with the '' default (bare builders return none)"
+    assert 'const { main, count, chunk = "", image = "" } =' in body, (
+        "the chunk + image parts destructure with the '' defaults"
+        " (bare builders return none)"
     )
-    i_set = body.find("setSyncLabel(main, count, chunk, true)")
+    i_set = body.find("setSyncLabel(main, count, chunk, true, image)")
     assert -1 < i_fall < i_set, "the parts are written after the split (running = true)"
     assert body.count("setSyncLabel(") == 1, "one writer call per entry"
     assert "syncBtn.title =" not in body, ("the title write lives in setSyncLabel (A4)")
@@ -713,6 +798,25 @@ enterSyncRunningState("sync", "src/a/b.md", 2, 3, {
   current_chunk: 14, chunks_total: 38,
 });
 show("import-inflight");
+// 1b. Phase 152 (task 03): import, a PDF's vision span in flight
+//     (the owner's stuck label — the last chunk position 125/126 is
+//     STILL on the status + the climbing image position 3/15): the
+//     composed full string carries BOTH parts (chunk keeps its span,
+//     the image part rides the string only — no span):
+enterSyncRunningState("sync", "src/a/b.md", 2, 3, {
+  phase: "import", current_file: "src/a/b.md", files_done: 2, files_total: 3,
+  current_chunk: 125, chunks_total: 126, images_done: 3, images_total: 15,
+});
+show("import-image-span");
+// 1c. Phase 152 (task 03): import, a purely pictorial PDF (no text
+//     chunks — the chunk position never left 0/0) in its vision
+//     span: the image part shows ALONE (the chunk part stays hidden
+//     + empty — the two parts are independent):
+enterSyncRunningState("sync", "src/a/b.md", 2, 3, {
+  phase: "import", current_file: "src/a/b.md", files_done: 2, files_total: 3,
+  current_chunk: 0, chunks_total: 0, images_done: 7, images_total: 7,
+});
+show("import-image-only");
 // 2. import before the first batch (0/0) — the chunk part hides:
 enterSyncRunningState("sync", "src/a/b.md", 2, 3, {
   phase: "import", current_file: "src/a/b.md", files_done: 2, files_total: 3,
@@ -893,6 +997,96 @@ test-document-viewer pattern). The pinned "· chunk n/m" part (the
         )
 
 
+@pytest.mark.skipif(not HAVE_NODE, reason="node not available")
+def test_image_part_renders_only_in_the_import_phase_under_node() -> None:
+    """Phase 152 (task 03): behavioral check (node) of the REAL label
+    path — the extracted fmtSyncPhaseLabel / setSyncLabel /
+    enterSyncRunningState run against the stub element refs. The
+    pinned "· image n/m" part (the task's fixture values):
+
+    * import, a PDF's vision span in flight (images_total 15 > 0)
+      WITH the last chunk position still on the status (125/126):
+      the composed full string carries BOTH parts — the box, the
+      button title, and the running aria-label all read exactly
+      "Syncing… src/a/b.md (2/3) · chunk 125/126 · image 3/15"
+      (the owner's stuck label, now moving); the chunk part KEEPS
+      its span ("· chunk 125/126" visible) and the image part gets
+      NO span of its own (it rides the composed string only — the
+      running button is the spinner, phase 149);
+    * import, a purely pictorial PDF (the chunk position never left
+      0/0) in its vision span: the image part shows ALONE —
+      "Syncing… src/a/b.md (2/3) · image 7/7" — and the chunk part
+      stays hidden + empty (the two parts are distinct and each
+      disappears independently on its own 0/0);
+    * every OTHER running state (the pre-batch import, the prelude,
+      the 202/409 click, the upload run, the overview, the
+      summaries — the status's image fields are 0/0 or cleared at
+      those transitions) leaves the composed string image-free.
+
+    The box / Cancel-control / is-running contracts of the running
+    state apply to the new rows exactly as to every other running
+    state (the harness's other pins cover them per tag)."""
+    rows = _chunk_harness_rows()
+    by_tag = {row[0]: row for row in rows}
+    # row = [tag, title, announcer (the box's text — the sole live
+    #        region, phase 150), main, count, countHidden, chunk,
+    #        chunkHidden, cancelHidden, cancelDisabled, progressText,
+    #        progressHidden, btnRunning, iconSpinning, ariaLabel]
+
+    # 1b. both parts at once — the composed full string, byte-for-byte
+    #     (the image part LAST, after the chunk part), on the box, the
+    #     title, and the running aria-label (A4) — the chunk span
+    #     keeps its own part.
+    full = "Syncing… src/a/b.md (2/3) · chunk 125/126 · image 3/15"
+    row = by_tag["import-image-span"]
+    assert row[1] == full, row[1]
+    assert row[2] == full, "the announcer (the box) hears the whole thing (A4)"
+    assert row[14] == full, "the running button's accessible name is the full label"
+    assert row[10] == full, "the detail box carries the EXACT full label"
+    assert row[6] == "· chunk 125/126" and row[7] is False, (
+        "the chunk part keeps its own span alongside the image part"
+    )
+
+    # 1c. the image part alone (the chunk position never left 0/0 —
+    #     the two parts are independent).
+    row = by_tag["import-image-only"]
+    assert row[1] == "Syncing… src/a/b.md (2/3) · image 7/7", row[1]
+    assert row[2] == row[1] and row[14] == row[1], row
+    assert row[6] == "" and row[7] is True, (
+        "the chunk part stays hidden + empty (its own 0/0)"
+    )
+
+    # The pre-batch import (0/0 on BOTH positions) and every other
+    # running state: the composed string is image-free (the part
+    # hides on its own 0/0 — the overview/summaries transitions
+    # clear the position server-side, the bare builders carry none).
+    for tag, expected_title in (
+        ("import-inflight", "Syncing… src/a/b.md (2/3) · chunk 14/38"),
+        ("import-prebatch", "Syncing… src/a/b.md (2/3)"),
+        ("prelude", "Syncing…"),
+        ("click", "Syncing…"),
+        ("upload", "Importing"),
+        ("overview", "Writing KB overview…"),
+        ("summaries", "Summarizing folders… src/alpha (2/3)"),
+    ):
+        row = by_tag[tag]
+        assert row[1] == expected_title, (tag, row)
+        assert "· image" not in row[1], (
+            f"{tag}: the image part shows only while images_total > 0"
+        )
+
+    # The running-state contracts apply to the new rows too: box
+    # visible + is-running + spinning, the Cancel control shown +
+    # enabled (a sync run in flight).
+    for tag in ("import-image-span", "import-image-only"):
+        row = by_tag[tag]
+        assert row[11] is False, f"{tag}: the box must be visible while in flight"
+        assert row[12] is True and row[13] is True, f"{tag}: is-running + spinning"
+        assert row[8] is False and row[9] is False, (
+            f"{tag}: the Cancel control is shown + enabled while a sync runs"
+        )
+
+
 @pytest.mark.skipif(not HAVE_NODE, reason="node not available")
 def test_progress_box_and_running_button_contract_under_node() -> None:
     """Phase 149 (task 01) — the detail box + the icon-only running
@@ -995,14 +1189,16 @@ def test_progress_box_and_running_button_contract_under_node() -> None:
 
 
 def test_every_other_state_passes_no_chunk_text() -> None:
-    """Source-level pin of the render-only-while-chunks_total>0
-    contract: the import fall-through is the ONLY place in sources.js
-    that reads the status's chunk fields (the summaries/overview
-    branches and the bare builders never touch them), and every
-    NON-running label emission passes no chunk text (the part hides):
-    the settle/idle labels go through setSyncLabel(label) — the only
-    running-flagged call site (phase 149: `running = true`) is the
-    running entry."""
+    """Source-level pin of the render-only-while-chunks_total>0 +
+    render-only-while-images_total>0 contracts: the import
+    fall-through is the ONLY place in sources.js that reads the
+    status's chunk OR image fields (the summaries/overview branches
+    and the bare builders never touch them), and every
+    NON-running label emission passes no chunk/image text (both
+    parts hide): the settle/idle labels go through setSyncLabel(label)
+    — the only running-flagged call site (phase 149: `running = true`)
+    is the running entry (phase 152: passing the image part as its
+    fifth argument)."""
     js = _js()
     body = _fn(js, "fmtSyncPhaseLabel")
     i_fall = body.find("fmtSyncLabel(")
@@ -1010,12 +1206,21 @@ def test_every_other_state_passes_no_chunk_text() -> None:
     assert "chunks_total" not in pre and "current_chunk" not in pre, (
         "the overview/summaries branches never read the chunk fields"
     )
+    assert "images_total" not in pre and "images_done" not in pre, (
+        "the overview/summaries branches never read the image fields (phase 152)"
+    )
     fall = body[i_fall:]
     assert "status.chunks_total" in fall and "status.current_chunk" in fall
-    assert js.count("setSyncLabel(main, count, chunk, true)") == 1, (
-        "the ONLY running-flagged call site is the running entry (phase 149)"
+    assert "status.images_total" in fall and "status.images_done" in fall, (
+        "the import fall-through is the ONLY reader of the image fields (phase 152)"
+    )
+    assert js.count("setSyncLabel(main, count, chunk, true, image)") == 1, (
+        "the ONLY running-flagged call site is the running entry (phase 149"
+        " — phase 152: with the image part)"
+    )
+    assert "setSyncLabel(label)" in js, (
+        "the settle/idle labels stay two-arg (no chunk/image)"
     )
-    assert "setSyncLabel(label)" in js, "the settle/idle labels stay two-arg (no chunk)"
 
 
 # ---------- the two-job tick (startSyncPolling) ----------

tests/unit/test_importer.py

@@ -18,7 +18,8 @@ import base64
 import hashlib
 import logging
 import os
-from collections.abc import Callable
+import textwrap
+from collections.abc import Callable, Sequence
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -41,6 +42,11 @@ from app.rag.importer import (
     match_extension,
 )
 from app.rag.llm import EmbeddingError, LLMError
+from app.rag.pdf import (
+    extract_pdf_images_with_stats,
+    extract_pdf_page_texts,
+    pictorial_pages,
+)
 from tests.fakes import FakeEmbedder
 
 #: The original seven A9 formats as dotted suffixes (pre-phase-47 default
@@ -2088,3 +2094,504 @@ def test_fallback_embedded_images_still_join_the_pipeline(
         assert image_doc.content != d1 and image_doc.content != d2
     finally:
         _cleanup_source(db, "both")
+
+
+# ---------- phase 152 (task 02): on_image_progress plumbing ----------
+
+#: A text block comfortably over the phase-140 100-char sparseness bar
+#: (a USABLE layer, zero pictorial candidates even with the toggle on
+#: — the "no vision span" class). Inserted wrapped at 60 chars: a
+#: single long line would be CLIPPED at the page edge on extraction,
+#: while the wrapped multi-line read round-trips (the phase-140
+#: fixture's class property).
+_P152_DENSE_TEXT = (
+    "The quadlet unit file defines the container interface, the image, "
+    "the network, and the restart policy. Every key maps to a docker "
+    "flag in the generated service file, and comments use the usual "
+    "semicolon form throughout the whole document."
+)
+assert len(_P152_DENSE_TEXT.strip()) > 100  # the class property, not a typo
+
+
+def _p152_solid_png(value: int) -> bytes:
+    """A 40x30 solid-color PNG (channel *value*): every value used
+    below stays under the phase-151 blank gate's 235 mean floor (kept,
+    never blank-skipped), and distinct values are distinct images (the
+    dedupe keeps every one)."""
+    pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 40, 30), 0)
+    pix.clear_with(value)
+    return pix.tobytes("png")
+
+
+def _build_image_pdf(
+    path: Path,
+    values: Sequence[int],
+    dense: bool = True,
+    repeat_first: bool = False,
+) -> None:
+    """A REAL one-page PDF: *dense* a text block over the 100-char bar
+    (a USABLE layer — the fallback never fires; zero pictorial
+    candidates even with the toggle on) carrying *values* the
+    solid-color embedded images (one placement each, distinct).
+    ``dense=False`` inserts the sparse ``BESTA`` marker instead (a
+    pictorial candidate under the default 100-char bar).
+    ``repeat_first`` adds a SECOND placement of the first image (the
+    same stream — one xref, the parser's repeated-placement class,
+    phase 151: the dedupe drops it, the position count is
+    post-dedupe)."""
+    doc = fitz.open()
+    page = doc.new_page()
+    text = _P152_DENSE_TEXT if dense else "BESTA"
+    for i, line in enumerate(textwrap.wrap(text, 60)):
+        page.insert_text((72, 72 + i * 14), line)
+    y = 300
+    for value in values:
+        page.insert_image(fitz.Rect(72, y, 112, y + 30), stream=_p152_solid_png(value))
+        y += 40
+    if repeat_first:
+        page.insert_image(
+            fitz.Rect(150, 300, 190, 330), stream=_p152_solid_png(values[0])
+        )
+    doc.save(path)
+    doc.close()
+
+
+def _build_pictorial_image_pdf(path: Path, values: Sequence[int]) -> None:
+    """A REAL two-page PDF — the ONE continuous counter's two-span
+    case: page 1 is SPARSE (``BESTA`` — a pictorial candidate under
+    the default 100-char bar), page 2 is dense (never a candidate) and
+    carries *values* the solid-color embedded images. Both the
+    phase-140 pictorial pass (before the document commit) and the
+    phase-126 embedded-image pass (after it) fire on it."""
+    doc = fitz.open()
+    doc.new_page().insert_text((72, 72), "BESTA")
+    page = doc.new_page()
+    for i, line in enumerate(textwrap.wrap(_P152_DENSE_TEXT, 60)):
+        page.insert_text((72, 72 + i * 14), line)
+    y = 300
+    for value in values:
+        page.insert_image(fitz.Rect(72, y, 112, y + 30), stream=_p152_solid_png(value))
+        y += 40
+    doc.save(path)
+    doc.close()
+
+
+def _build_garbage_image_pdf(path: Path, values: Sequence[int]) -> None:
+    """A REAL two-page garbage-text PDF (every page XML-ish — the
+    phase-133 gate fires, the page-screenshot fallback runs) with
+    *values* the solid-color embedded images on page 2 — the fallback
+    span + the embedded span, ONE continuous counter."""
+    doc = fitz.open()
+    for p in range(1, 3):
+        page = doc.new_page()
+        for i in range(10):
+            page.insert_text((72, 72 + i * 14), f"<foo:p{p}>{i}</foo:p{p}>")
+    y = 400
+    for value in values:
+        doc[1].insert_image(
+            fitz.Rect(72, y, 112, y + 30), stream=_p152_solid_png(value)
+        )
+        y += 40
+    doc.save(path)
+    doc.close()
+
+
+def _p152_summary_fields(s: ImportSummary) -> tuple:
+    """One run's observable outcome (the byte-identity pin's left
+    side): every count a hook could plausibly move + the changed
+    paths' RELS (the source names are equal by construction — the two
+    roots carry the same directory name)."""
+    return (
+        s.files, s.added, s.updated, s.unchanged, s.pruned, s.errors,
+        s.chunks, s.embed_batches, s.summaries, s.summary_errors,
+        s.images_failed, s.pdfs_screenshotted, s.pdfs_pictorial,
+        s.pdf_pictorial_pages, s.pdf_images_deduped, s.pdf_images_blank_skipped,
+        tuple(rel for _, rel in s.changed_paths),
+    )
+
+
+def _p152_rows(db, source: str) -> list[tuple]:
+    """A source's document rows (the byte-identity pin's right side),
+    identity-blind: the fields a hook could plausibly move."""
+    docs = db.scalars(select(Document).where(Document.source == source)).all()
+    return sorted(
+        (d.path, d.title, d.content, d.content_hash, d.is_image, d.summary,
+         d.image_context)
+        for d in docs
+    )
+
+
+
+def test_on_image_progress_embedded_pass_climbs_one_per_image(
+    db, tmp_path: Path
+) -> None:
+    """The phase-152 core pin: a PDF with N DISTINCT embedded images
+    (one placement each — post-dedupe, post-blank-gate: the work
+    actually about to happen) fires the hook ONCE per image, 1-based,
+    with the span's total, at the START of each image's pipeline step:
+    ``(1, N) … (N, N)``. A repeated placement of the first image (the
+    parser's dedupe, phase 151) adds NO position (the total is the
+    post-dedupe count), a clean usable text layer burns no fallback or
+    pictorial span (the embedded pass's totals stand alone), and the
+    phase-147 fast path on a re-sync (raw bytes unchanged) burns no
+    vision call and fires the hook never."""
+    root = tmp_path / "imgprog"
+    root.mkdir()
+    values = (90, 130, 170, 210)  # 4 distinct images, all under the blank floor
+    _build_image_pdf(root / "report.pdf", values, repeat_first=True)
+    n = len(values)
+    llm = _PageVisionLLM()
+    llm.settings = _pdf_settings(tmp_path, images=True)
+    events: list[tuple[int, int]] = []
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        # The PDF + its N image docs; the repeat placement deduped.
+        assert (summary.files, summary.added) == (1, 1 + n)
+        assert summary.pdf_images_deduped == 1
+        assert (summary.images_failed, summary.pdfs_screenshotted,
+                summary.pdfs_pictorial) == (0, 0, 0)
+        # The embedded pass is the ONLY vision span: N slow calls.
+        assert llm.vision_calls == n
+        # The core pin: once per image, 1-based, the span's total.
+        assert events == [(i, n) for i in range(1, n + 1)]
+        # The phase-147 fast path: a re-sync (bytes unchanged) burns no
+        # vision call and fires the hook never — the events list grows
+        # not at all.
+        summary2 = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        assert summary2.unchanged == 1
+        assert llm.vision_calls == n
+        assert len(events) == n
+    finally:
+        _cleanup_source(db, root.name)
+
+
+def test_on_image_progress_pictorial_then_embedded_is_one_counter(
+    db, tmp_path: Path
+) -> None:
+    """ASSUMPTION A3: the spans of ONE file are ONE continuous counter
+    — the embedded pass starts where the pictorial pass ended (``base
+    = images_done_so_far``): the positions climb ``1…C+N`` with NO
+    restart and never above the combined total (the pictorial span's
+    total is its candidate-page count, the embedded span's total the
+    combined count)."""
+    root = tmp_path / "imgprog-two"
+    root.mkdir()
+    values = (90, 130, 170)
+    _build_pictorial_image_pdf(root / "ikea.pdf", values)
+    # The span inputs, computed IN the test (no hardcoded lists — the
+    # phase-140 idiom): the deterministic gate's candidates and the
+    # post-dedupe/post-gate image count.
+    pdf = root / "ikea.pdf"
+    page_texts = extract_pdf_page_texts(pdf)
+    n_cand = len(pictorial_pages(page_texts, min_chars=100))
+    images, _ = extract_pdf_images_with_stats(pdf)
+    n_img = len(images)
+    assert (n_cand, n_img) == (1, len(values))
+    llm = _PageVisionLLM()
+    llm.settings = Settings(
+        _env_file=None,  # pyright: ignore[reportCallIssue]
+        import_extensions="md,txt,pdf",
+        images=True,
+        image_dir=str(tmp_path / "images"),
+        pdf_pictorial=True,  # default on — the pass runs on the sparse page
+    )
+    events: list[tuple[int, int]] = []
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        assert (summary.files, summary.added) == (1, 1 + n_img)
+        assert (summary.pdfs_pictorial, summary.pdf_pictorial_pages) == (1, 1)
+        # The continuous counter: 1…C (total C) then C+1…C+N (total
+        # C+N) — no restart, never above the combined total.
+        assert events == (
+            [(i, n_cand) for i in range(1, n_cand + 1)]
+            + [(n_cand + i, n_cand + n_img) for i in range(1, n_img + 1)]
+        )
+    finally:
+        _cleanup_source(db, root.name)
+
+
+def test_on_image_progress_fallback_span_climbs_into_embedded(
+    db, tmp_path: Path
+) -> None:
+    """The owner's stuck-label class: a garbage-text PDF's per-page
+    screenshot describes (the phase-133 fallback span — the span that
+    froze the label for minutes) report the rendered-page count
+    (``1…M`` with total ``M``) and hand their reported count to the
+    embedded pass, which continues the SAME counter (``M+1`` with
+    total ``M+N``) — every slow call has a position before it."""
+    root = tmp_path / "imgprog-fb"
+    root.mkdir()
+    _build_garbage_image_pdf(root / "garbage.pdf", (90,))
+    llm = _PageVisionLLM()
+    llm.settings = _pdf_settings(tmp_path, images=True)
+    events: list[tuple[int, int]] = []
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        assert summary.pdfs_screenshotted == 1
+        assert (summary.files, summary.added, summary.images_failed) == (1, 2, 0)
+        # 2 fallback pages (total 2) + 1 embedded image continuing at
+        # 3 (total 3) — ONE counter across the two spans.
+        assert events == [(1, 2), (2, 2), (3, 3)]
+        # Every slow call (2 pages + 1 image) was preceded by a
+        # position — the label never sits frozen on a busy run.
+        assert llm.vision_calls == 3
+    finally:
+        _cleanup_source(db, root.name)
+
+
+def test_on_image_progress_text_only_pdf_never_fires(
+    db, tmp_path: Path
+) -> None:
+    """A clean-text PDF with NO embedded images fires the hook NEVER,
+    with BOTH default-on toggles: the garbage gate stays closed (no
+    fallback span), the pictorial pass's deterministic gate finds zero
+    candidates (a no-op — the clean-text byte-identity class), and the
+    embedded pass's list is empty. The skipped spans report nothing
+    and advance the counter not at all (there is no vision call at
+    all)."""
+    root = tmp_path / "imgprog-none"
+    root.mkdir()
+    _build_image_pdf(root / "clean.pdf", ())  # dense text, zero images
+    llm = _PageVisionLLM()
+    llm.settings = Settings(
+        _env_file=None,  # pyright: ignore[reportCallIssue]
+        import_extensions="md,txt,pdf",
+        images=True,
+        image_dir=str(tmp_path / "images"),
+        pdf_pictorial=True,  # default on — the pass runs and no-ops
+    )
+    events: list[tuple[int, int]] = []
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        assert (summary.files, summary.added) == (1, 1)
+        assert (summary.pdfs_screenshotted, summary.pdfs_pictorial) == (0, 0)
+        assert llm.vision_calls == 0  # no span ran — zero slow calls
+        assert events == []
+    finally:
+        _cleanup_source(db, root.name)
+
+
+def test_on_image_progress_toggle_off_reports_nothing(
+    db, tmp_path: Path
+) -> None:
+    """The ``images`` toggle OFF skips the embedded span entirely (the
+    prune guard keeps any existing image docs) — a PDF that WOULD
+    carry N embedded images reports nothing (a skipped span advances
+    the counter not at all) and indexes as text only."""
+    root = tmp_path / "imgprog-off"
+    root.mkdir()
+    _build_image_pdf(root / "off.pdf", (90, 130))
+    llm = _PageVisionLLM()
+    llm.settings = _pdf_settings(tmp_path, images=False)
+    events: list[tuple[int, int]] = []
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        assert (summary.files, summary.added) == (1, 1)
+        assert llm.vision_calls == 0
+        assert events == []
+    finally:
+        _cleanup_source(db, root.name)
+
+
+def test_on_image_progress_none_run_is_byte_identical(
+    db, tmp_path: Path
+) -> None:
+    """The regression pin: a hook-less run (the CLI/chat default —
+    ``on_image_progress=None``, every call site a guarded no-op) lands
+    byte-identical to a run WITH the hook: the same documents (path,
+    title, content, digest, image flag, summary, context) and the same
+    summary counters. The hook observes; it never changes the run."""
+    # Both roots carry the SAME directory name — the source (the
+    # walk's root.name) is the document's provenance and part of the
+    # stored image context, so the comparison needs it equal.
+    root_a = tmp_path / "a" / "same_src"
+    root_a.mkdir(parents=True)
+    root_b = tmp_path / "b" / "same_src"
+    root_b.mkdir(parents=True)
+    values = (90, 130, 170, 210)
+    _build_image_pdf(root_a / "same.pdf", values)
+    _build_image_pdf(root_b / "same.pdf", values)
+    llm_a, llm_b = _PageVisionLLM(), _PageVisionLLM()
+    llm_a.settings = _pdf_settings(tmp_path, images=True)
+    llm_b.settings = _pdf_settings(tmp_path, images=True)
+    events: list[tuple[int, int]] = []
+    try:
+        # Run A (hook-less) first, snapshot its rows, clear the shared
+        # source name, then run B (hooked) — the rows line up 1:1.
+        summary_a = asyncio.run(import_sources([root_a], llm_a, session=db))
+        rows_a = _p152_rows(db, "same_src")
+        _cleanup_source(db, "same_src")
+        summary_b = asyncio.run(
+            import_sources(
+                [root_b], llm_b, session=db,
+                on_image_progress=lambda pos, total: events.append((pos, total)),
+            )
+        )
+        rows_b = _p152_rows(db, "same_src")
+        # The hook ran in run B (N image positions, the span's total)
+        # — and changed nothing observable.
+        assert events == [(i, len(values)) for i in range(1, len(values) + 1)]
+        assert _p152_summary_fields(summary_a) == _p152_summary_fields(summary_b)
+        assert rows_a == rows_b
+    finally:
+        _cleanup_source(db, "same_src")
+
+
+class _InterleavingVisionLLM(_PageVisionLLM):
+    """A ``_PageVisionLLM`` variant that records each vision call in
+    a SHARED event sequence (phase 152, task 04): the importer fires
+    the ``on_image_progress`` hook BEFORE the
+    ``_index_one_embedded_image`` describe seam, so the sequence reads
+    hook, vision, hook, vision, … — the interleaving pin (the label
+    moves at the START of each slow call, never after it)."""
+
+    def __init__(self, events: list[tuple]) -> None:
+        super().__init__()
+        self.events = events
+
+    async def chat(self, messages, model=None):  # pyright: ignore[reportIncompatibleMethodOverride]
+        reply = await super().chat(messages, model)
+        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
+        if isinstance(user, list):
+            self.events.append(("vision", self.vision_calls))
+        return reply
+
+
+def test_on_image_progress_hook_fires_before_each_describe_call(
+    db, tmp_path: Path
+) -> None:
+    """Phase 152 (task 04): the hook fires BEFORE the corresponding
+    describe call (the contract: at the START of each slow call, never
+    after it) — the interleaving pinned by recording events in the
+    fake: for a PDF with N embedded images the sequence reads
+    hook(1,N), vision(1), hook(2,N), vision(2), …, hook(N,N),
+    vision(N) (the fake's vision call IS the describe seam the hook
+    precedes — the label moves while the describe is about to start,
+    so a busy run never reads frozen)."""
+    root = tmp_path / "imgprog-order"
+    root.mkdir()
+    values = (90, 130, 170)
+    _build_image_pdf(root / "order.pdf", values)
+    n = len(values)
+    events: list[tuple] = []
+    llm = _InterleavingVisionLLM(events)
+    llm.settings = _pdf_settings(tmp_path, images=True)
+    try:
+        summary = asyncio.run(
+            import_sources(
+                [root], llm, session=db,
+                on_image_progress=lambda pos, total: events.append(("image", pos, total)),
+            )
+        )
+        assert (summary.files, summary.added) == (1, 1 + n)
+        assert llm.vision_calls == n  # one describe per image
+        # The interleaving: hook BEFORE each describe call, in order.
+        expected: list[tuple] = []
+        for i in range(1, n + 1):
+            expected.append(("image", i, n))
+            expected.append(("vision", i))
+        assert events == expected
+    finally:
+        _cleanup_source(db, root.name)
+
+
+def test_on_image_progress_blank_gate_excludes_skipped_images_from_the_total(
+    db, tmp_path: Path
+) -> None:
+    """Phase 152 (task 04, the phase-151 interaction): images skipped
+    by the dedupe/blank gate are NOT counted in the hook's total —
+    the total is the work actually about to happen (the pass's
+    post-dedupe, post-gate list). A PDF with 2 real + 1 near-white
+    image (250 — the blank gate's class, mean ≥ 235): the gate
+    (default ON) drops the near-white one BEFORE the pass — the hook
+    fires for the 2 kept with total 2 (the skipped image burns no
+    vision call, no position); with the kill switch OFF
+    (``pdf_image_skip_blank=False`` — the pre-phase-151 behavior) the
+    near-white image is indexed too — the hook fires for all 3 with
+    total 3 (the total is the work about to happen under the knob's
+    state)."""
+    values_a = (90, 130, 250)   # 250 = near-white (the blank gate's class)
+    values_b = (110, 150, 250)  # distinct bytes — a different document set
+    # Both roots carry the SAME directory name — the source is part of
+    # the stored image context, so the runs line up 1:1 (the house
+    # cleanup idiom separates them).
+    root_a = tmp_path / "a" / "same_src"
+    root_a.mkdir(parents=True)
+    root_b = tmp_path / "b" / "same_src"
+    root_b.mkdir(parents=True)
+    _build_image_pdf(root_a / "blank.pdf", values_a)
+    _build_image_pdf(root_b / "blank.pdf", values_b)
+    llm_a = _PageVisionLLM()
+    llm_a.settings = _pdf_settings(tmp_path, images=True)  # skip_blank default ON
+    events_a: list[tuple[int, int]] = []
+    try:
+        summary_a = asyncio.run(
+            import_sources(
+                [root_a], llm_a, session=db,
+                on_image_progress=lambda pos, total: events_a.append((pos, total)),
+            )
+        )
+        # The blank gate dropped the near-white image BEFORE the
+        # pass — the total is the 2 actually about to happen.
+        assert summary_a.pdf_images_blank_skipped == 1
+        assert (summary_a.files, summary_a.added) == (1, 3)
+        assert llm_a.vision_calls == 2
+        assert events_a == [(1, 2), (2, 2)]
+        # The knob OFF: the near-white image is indexed too — the
+        # total is all 3 (the pre-phase behavior, the work about to
+        # happen under the knob's state). Fresh source (the house
+        # cleanup idiom — run B's rows would otherwise be updates of
+        # run A's rows, same synthetic image paths).
+        _cleanup_source(db, "same_src")
+        llm_b = _PageVisionLLM()
+        llm_b.settings = Settings(
+            _env_file=None,  # pyright: ignore[reportCallIssue]
+            import_extensions="md,txt,pdf",
+            images=True,
+            image_dir=str(tmp_path / "images"),
+            pdf_pictorial=False,
+            pdf_image_skip_blank=False,
+        )
+        events_b: list[tuple[int, int]] = []
+        summary_b = asyncio.run(
+            import_sources(
+                [root_b], llm_b, session=db,
+                on_image_progress=lambda pos, total: events_b.append((pos, total)),
+            )
+        )
+        assert summary_b.pdf_images_blank_skipped == 0
+        assert (summary_b.files, summary_b.added) == (1, 4)
+        assert llm_b.vision_calls == 3
+        assert events_b == [(1, 3), (2, 3), (3, 3)]
+    finally:
+        _cleanup_source(db, "same_src")

tests/unit/test_sync_button.py

@@ -110,6 +110,56 @@ poll tree's new 5th branch (after failed, before the upload branches
 — the 7-branch tree is now 8), the neutral ``applySyncCancelled``
 (no modal, no banner, no error styling), and the
 ``initSyncButton`` re-attach of a ``cancelled`` run.
+
+Phase 152 (task 01): the status contract gains the in-flight file's
+VISION-image position — ``images_done`` (the 1-based index of the
+image about to be described) + ``images_total`` (the file's total
+images for this run — the three PDF vision spans share ONE continuous
+counter per file, the importer's ``on_image_progress`` hook that
+task 02/03 wire). The idle dict pins them 0/0; the per-file reset is
+the pairing rule, extended — a ``progress`` call names the file AND
+clears the image position, pinned through the runner's own ``_hook``
+closure (the two-file fake below: the position fired through the
+runner's own ``_image_hook`` closure while file A is in flight is
+0/0 when file B is named — a file with no images must not inherit
+the previous PDF's position); the clear rule mirrors the chunk
+position — 0/0 at the overview/summaries transitions and in EVERY
+terminal (success, failed, and cancelled — each pinned with a live
+position in flight; task 04 moves every live position below onto the
+runner's OWN closures, the phase-64/129 idiom), while the FILE
+counts keep their final values (the convention split: a
+sub-position of a file that's done clears with the file).
+
+Phase 152 (task 03): the runner's THIRD paired closure —
+``_image_hook`` (beside ``_hook`` and ``_chunk_hook``) — lands the
+importer's ``on_image_progress`` pairs on the status, and the import
+call now passes ``on_image_progress=_image_hook`` (the wiring pin:
+the fakes receive the kwarg and record the live closure). The
+frontend gains the "· image n/m" label PART: the running entry
+destructures ``image = ""`` and the writer call becomes
+``setSyncLabel(main, count, chunk, true, image)`` — the image part
+rides ONLY the composed full string (box + title + running
+aria-label — no new button span; the running button is the
+spinner), the chunk part keeps its own span, and the bare builders
++ non-running calls pass nothing (both parts default "").
+
+Phase 152 (task 04): the position pins move onto the runner's OWN
+closures (the phase-64/129 idiom — the fakes fire the image pairs
+through the ``on_image_progress`` kwarg, the ``_image_hook`` under
+test, instead of standing in on the status object): a mid-run pair
+(3/15, then 15/15) climbs on the status BETWEEN ticks while
+``current_file`` and the chunk position keep their own values (the
+THREE positions are independent sub-positions of the file), the
+per-file reset, the overview/summaries transition clears, and the
+FAILED + CANCELLED terminals all fire their live positions through
+the closure too. The frontend source pins the ``fmtSyncPhaseLabel``
+image part (present iff ``images_total > 0`` — the exact guard +
+template), the ``overview`` / ``summaries`` branches' ``image: ""``
+(unchanged labels), and ``setSyncLabel``'s composed ``full`` string
+byte-for-byte (a status carrying both positions reads "… (2540/2540)
+· chunk 125/126 · image 3/15" on the box + title + running
+aria-label — the runtime string is behaviorally pinned in
+tests/unit/test_frontend_sync_upload.py's node check).
 """
 from __future__ import annotations
 
@@ -393,13 +443,16 @@ def test_sources_js_running_state_is_never_stale() -> None:
     Phase 128 (task 01): the entry hands the label PARTS (the
     phase-aware split — sync + status → fmtSyncPhaseLabel, the
     fall-through → the byte-identical fmtSyncLabel) to the SINGLE
-    writer, setSyncLabel(main, count, chunk — phase 129: the chunk
-    part destructures with a "" default), which lands the main part,
+    writer, setSyncLabel(main, count, chunk, running, image — phase
+    129: the chunk part destructures with a "" default, phase 152:
+    the image part does too), which lands the main part,
     the pinned count part (hidden when absent), the pinned chunk part
     (phase 129 — hidden when absent), and the FULL untruncated label
-    on the button title + #sync-progress (A4) — the direct span /
-    title / announcer writes are GONE from the entry (they live in
-    setSyncLabel + setSyncProgress). Phase 149 (task 01): the writer call passes
+    on the button title + #sync-progress (A4 — phase 152: with the
+    image part appended after the chunk part when a PDF's vision span
+    is in flight; the image part gets no span of its own) — the
+    direct span / title / announcer writes are GONE from the entry
+    (they live in setSyncLabel + setSyncProgress). Phase 149 (task 01): the writer call passes
     `running = true` (the detail box reveals with the full label),
     and the button's accessible name rides in the writer too (the
     full live label — the icon-only running button keeps a name);
@@ -414,8 +467,10 @@ def test_sources_js_running_state_is_never_stale() -> None:
     assert "fmtSyncLabel(kind, currentFile, done, total)" in body
     # …and the single writer lands the parts (A4 in it — the title is
     # set for EVERY running label: the file-only rule is adjusted
-    # away) with the phase-149 `running` flag (the box reveals).
-    assert "setSyncLabel(main, count, chunk, true)" in body
+    # away) with the phase-149 `running` flag (the box reveals) +
+    # the phase-152 image part (rides only the composed full string —
+    # no span).
+    assert "setSyncLabel(main, count, chunk, true, image)" in body
     assert body.count("setSyncLabel(") == 1, "the entry writes ONLY through the parts writer"
     assert "syncBtn.title =" not in body, ("the title write lives in setSyncLabel (A4)")
     assert "syncLabel.textContent" not in body, ("no direct span write — the parts")
@@ -433,6 +488,70 @@ def test_sources_js_running_state_is_never_stale() -> None:
     )
 
 
+def test_fmt_sync_phase_label_image_part_is_pinned_source_level() -> None:
+    """Phase 152 (task 04): the frontend SOURCE pins the "· image
+    n/m" label part (the browser behavior is pinned in
+    tests/unit/test_frontend_sync_upload.py's node check — the REAL
+    extracted functions run against stub refs; this pin keeps the
+    template itself from drifting without a browser):
+
+    * the import fall-through is the ONLY reader of the status's
+      image fields, guarded on ``images_total > 0`` (the part is
+      present IFF ``images_total > 0`` — 0/0 hides it, the exact
+      guard + template + empty false-branch);
+    * the ``overview`` / ``summaries`` branches return ``image: ""``
+      (the position belongs to the import phase only — those labels
+      stay unchanged by the phase);
+    * the running entry destructures the part with a `""` default and
+      hands it to the single writer as the FIFTH argument (after the
+      `running` flag);
+    * ``setSyncLabel``'s composed ``full`` string appends the image
+      part LAST (after the chunk part) — so a status carrying both
+      positions reads exactly "Syncing… <file> (2540/2540) · chunk
+      125/126 · image 3/15" on the box, the title, and the running
+      aria-label (byte-for-byte; the docstring's worked example
+      carries the shape).
+    """
+    js = _text(SOURCES_JS)
+    builder = _body(js, "fmtSyncPhaseLabel")
+    # The import fall-through: present IFF images_total > 0 — the
+    # exact guard + template (the phase-129 chunk part's sibling,
+    # same shape).
+    assert (
+        "status.images_total > 0\n"
+        "        ? `· image ${status.images_done}/${status.images_total}`\n"
+        "        : \"\";"
+    ) in builder
+    assert "return { main, count, chunk, image };" in builder
+    # The overview branch: the unchanged label + the part absent.
+    assert (
+        'return { main: "Writing KB overview…", count: "", chunk: "", image: "" };'
+    ) in builder
+    # The summaries branch: the unchanged label + the part absent
+    # (the phase's own (n/m) is its position).
+    assert "image: \"\", // the image position belongs to the import phase only" in builder
+    assert "`Summarizing folders…${folder}`" in builder
+    # The running entry: the part destructures with a `""` default
+    # and rides the writer call as the fifth argument.
+    entry = _body(js, "enterSyncRunningState")
+    assert 'const { main, count, chunk = "", image = "" }' in entry
+    assert "setSyncLabel(main, count, chunk, true, image)" in entry
+    # The writer: the composed full string — the image part LAST,
+    # after the chunk part (the "… · chunk 125/126 · image 3/15"
+    # shape the docstring's worked example carries).
+    writer = _body(js, "setSyncLabel")
+    assert (
+        "const full =\n"
+        "      mainText +\n"
+        "      (countText ? ` ${countText}` : \"\") +\n"
+        "      (chunkText ? ` ${chunkText}` : \"\") +\n"
+        "      (imageText ? ` ${imageText}` : \"\");"
+    ) in writer
+    assert "· chunk 125/126 · image 3/15" in js, (
+        "the worked example (both positions at once) is documented"
+    )
+
+
 def test_sources_js_terminal_states() -> None:
     """Terminal rendering: success → enabled + 'Synced HH:MM' (local
     time of finished_at); failed → enabled + retry-ready 'Sync sources'
@@ -927,7 +1046,20 @@ class _GatedImport:
     ``_chunk_hook`` closure: pair one lands before the main park
     (readable mid-run), and every later pair lands after the main
     park + its own park (``chunk_release``), so each position is
-    readable on the status between ticks."""
+    readable on the status between ticks.
+
+    Phase 152 (task 03): the runner now ALWAYS passes the
+    image-position closure (task 03's wiring) — recorded for the
+    wiring pins.
+
+    Phase 152 (task 04): optionally ALSO fires the in-flight file's
+    image-position pairs (``image_pairs``) through the runner's OWN
+    ``_image_hook`` closure (pair one before the main park —
+    readable mid-run; every later pair after the main park + its own
+    park ``image_release`` — the phase-129 chunk idiom, mirrored for
+    the vision-image position). With no pairs the fake never fires
+    it (the status image fields stay 0/0, like the chunk fields of a
+    no-batch import)."""
 
     def __init__(
         self,
@@ -937,6 +1069,8 @@ class _GatedImport:
         fail: BaseException | None = None,
         chunk_pairs: Sequence[tuple[int, int]] = (),
         chunk_release: threading.Event | None = None,
+        image_pairs: Sequence[tuple[int, int]] = (),  # phase 152 (task 04)
+        image_release: threading.Event | None = None,  # phase 152 (task 04)
     ) -> None:
         self.summary = summary
         self.started = started
@@ -947,6 +1081,12 @@ class _GatedImport:
         self.chunk_pairs = list(chunk_pairs)
         self.chunk_release = chunk_release
         self.chunk_hook_calls: list[tuple[int, int]] = []
+        # Phase 152 (task 03): the image-position hook the runner
+        # passes (recorded for the wiring pins).
+        self.image_hook: Callable[[int, int], None] | None = None
+        self.image_pairs = list(image_pairs)
+        self.image_release = image_release
+        self.image_hook_calls: list[tuple[int, int]] = []
 
     async def __call__(
         self,
@@ -958,6 +1098,7 @@ class _GatedImport:
         session: object = None,
         progress: Callable[[str, str, int, int], None] | None = None,
         on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
+        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
         ignore_by_root: dict[str, list[str]] | None = None,  # phase 89
         include_hidden_by_root: dict[str, bool] | None = None,  # phase 105
         doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,  # phase 106
@@ -966,6 +1107,16 @@ class _GatedImport:
         if progress is not None:
             progress("repo", "notes/deep.md", 1, 3)
             self.hook_calls.append(("repo", "notes/deep.md", 1, 3))
+        # Phase 152 (task 03): the runner now ALWAYS passes the
+        # image-position closure (task 03's wiring) — recorded for the
+        # wiring pins.
+        self.image_hook = on_image_progress
+        # Phase 152 (task 04): image pair one lands BEFORE the main
+        # park (readable mid-run — the phase-129 chunk-pair-one
+        # idiom, mirrored for the vision-image position).
+        if on_image_progress is not None and self.image_pairs:
+            on_image_progress(*self.image_pairs[0])
+            self.image_hook_calls.append(self.image_pairs[0])
         if on_chunk_progress is not None and self.chunk_pairs:
             # Phase 129 (task 03): the file's chunk-position pairs,
             # fired through the runner's OWN hook closure (the closure
@@ -986,6 +1137,17 @@ class _GatedImport:
         else:
             self.started.set()
             await asyncio.to_thread(self.release.wait)  # park without freezing the worker loop
+        # Phase 152 (task 04): every later image pair lands after the
+        # main park + its own park (``image_release``), so each
+        # position is readable on the status between ticks. A wiring
+        # regression to None skips this branch and the pair tests
+        # time out.
+        if on_image_progress is not None:
+            for pair in self.image_pairs[1:]:
+                on_image_progress(*pair)
+                self.image_hook_calls.append(pair)
+                if self.image_release is not None:
+                    await asyncio.to_thread(self.image_release.wait)
         if self.fail is not None:
             raise self.fail
         return self.summary
@@ -1088,10 +1250,10 @@ def test_idle_status_pins_full_shape_including_progress_keys(
     fresh_sync_status: None,
 ) -> None:
     """Idle: the three phase-64 progress keys ride along as null/0/0,
-    the two phase-129 chunk keys as 0/0, the four phase-98 phase keys
-    as null/0/0/0, and EVERY pre-existing key is unchanged — the full
-    response dict is pinned, so the current UI and every existing
-    consumer keep working."""
+    the two phase-129 chunk keys as 0/0, the two phase-152 image keys
+    as 0/0, the four phase-98 phase keys as null/0/0/0, and EVERY
+    pre-existing key is unchanged — the full response dict is pinned,
+    so the current UI and every existing consumer keep working."""
     assert sync_api.sync_status() == {
         "state": "idle",
         "started_at": None,
@@ -1105,6 +1267,10 @@ def test_idle_status_pins_full_shape_including_progress_keys(
         # (the dataclass defaults, no special-casing).
         "current_chunk": 0,
         "chunks_total": 0,
+        # Phase 152 (task 01): the image-position keys — 0/0 idle
+        # (the dataclass defaults, no special-casing).
+        "images_done": 0,
+        "images_total": 0,
         # Phase 98 (task 01): the phase-machine keys — null/0/0/0 idle.
         "phase": None,
         "current_summary": None,
@@ -1145,13 +1311,19 @@ def test_mid_run_status_reports_current_file(
         assert s["files_done"] == 0 and s["files_total"] == 0
         assert s["summaries_done"] == 0 and s["summaries_total"] == 0
         assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+        # Phase 152 (task 01): the image position is 0/0 in the
+        # prelude too (no vision span yet).
+        assert s["images_done"] == 0 and s["images_total"] == 0
         clone.release.set()
         assert fake_import.started.wait(5.0), "the run never reached the import"
         # Import phase: the hook's file is live on the status, and the
         # phase machine names the phase (the summary span has not
         # started — its counters are still 0/0); no embed batch has
         # fired the chunk hook (this canned import passes no pairs),
-        # so the chunk position stays 0/0.
+        # so the chunk position stays 0/0; the image position stays
+        # 0/0 too (this canned import fires no image hook — task 03
+        # wires the runner's image closure, the importer's span is
+        # what would feed it).
         s = sync_api.sync_status()
         assert s["state"] == "running"
         assert s["phase"] == "import"
@@ -1161,6 +1333,7 @@ def test_mid_run_status_reports_current_file(
         assert s["current_summary"] is None
         assert s["summaries_done"] == 0 and s["summaries_total"] == 0
         assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+        assert s["images_done"] == 0 and s["images_total"] == 0
         fake_import.release.set()
     finally:
         clone.release.set()
@@ -1178,7 +1351,8 @@ def test_mid_run_status_reports_current_file(
     # folder step never fired the summary hook, so the summary counts
     # stay 0/0 (the keep-final-counts convention keeps what fired);
     # the chunk position clears with the file (phase 129 — it is a
-    # sub-position of a done file, never a keep-final counter).
+    # sub-position of a done file, never a keep-final counter), and
+    # so does the image position (phase 152 — same convention).
     s = sync_api.sync_status()
     assert s["state"] == "success"
     assert s["phase"] is None
@@ -1187,6 +1361,7 @@ def test_mid_run_status_reports_current_file(
     assert s["files_done"] == 1 and s["files_total"] == 3
     assert s["summaries_done"] == 0 and s["summaries_total"] == 0
     assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+    assert s["images_done"] == 0 and s["images_total"] == 0
     assert s["error"] is None
     assert s["started_at"] is not None and s["finished_at"] is not None
 
@@ -1215,6 +1390,7 @@ def test_failed_terminal_clears_current_file_keeps_counts(
             "failed: connection refused"
         ),
         chunk_pairs=[(1, 38)],  # phase 129: the live pair before the raise
+        image_pairs=[(3, 15)],  # phase 152 (task 04): the live pair before the raise
     )
     _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)
 
@@ -1231,6 +1407,13 @@ def test_failed_terminal_clears_current_file_keeps_counts(
         assert s["current_file"] == "repo/notes/deep.md"
         assert s["files_done"] == 1 and s["files_total"] == 3
         assert s["current_chunk"] == 1 and s["chunks_total"] == 38
+        # Phase 152 (task 04): a live image position in flight when
+        # the run dies — fired through the runner's OWN _image_hook
+        # closure (image_pairs[0] — the importer's span would do the
+        # same); the failed terminal must clear it (the
+        # sub-position of an abandoned file, like the chunk
+        # position).
+        assert s["images_done"] == 3 and s["images_total"] == 15
         fake_import.release.set()
     finally:
         clone.release.set()
@@ -1238,8 +1421,9 @@ def test_failed_terminal_clears_current_file_keeps_counts(
         thread.join(10.0)
     assert not thread.is_alive()
     assert errors == []
-    # Wiring: the pair fired through the runner's own closure.
+    # Wiring: the pairs fired through the runner's OWN closure.
     assert fake_import.chunk_hook_calls == [(1, 38)]
+    assert fake_import.image_hook_calls == [(3, 15)]
     s = sync_api.sync_status()
     assert s["state"] == "failed"
     assert s["current_file"] is None  # cleared in the terminal state
@@ -1251,6 +1435,11 @@ def test_failed_terminal_clears_current_file_keeps_counts(
     # terminal too — the convention split, pinned (the FILE counts
     # above keep their final values, the sub-position does not).
     assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+    # Phase 152 (task 04): the image position clears in the FAILED
+    # terminal too — the convention split, pinned (the live 3/15 fired
+    # through the closure mid-run is gone, the FILE counts keep their
+    # final values).
+    assert s["images_done"] == 0 and s["images_total"] == 0
     assert s["detail"] == {}
     error = s["error"] or ""
     assert "*****@aipi.example.com" in error  # credentials masked
@@ -1468,6 +1657,357 @@ def test_chunk_position_clears_at_phase_transitions_and_terminal(
     assert s["files_done"] == 1 and s["files_total"] == 3
 
 
+# ---------- phase 152 (task 01 + task 04): the in-flight file's -----
+# ---------- vision-image position (images_done / images_total) -------
+#
+# Same seams as the phase-129 block above. Task 04 pins every live
+# position through the runner's OWN closures (the phase-64/129 idiom):
+# the fakes fire the image pairs through the ``on_image_progress``
+# kwarg — the ``_image_hook`` under test (task 03's wiring: the
+# importer's span's exact path) — and drive the per-file ``_hook``
+# reset (through the ``progress`` kwarg the fake import receives) and
+# the transition/terminal clears.
+
+
+def test_mid_run_status_reports_image_position_between_ticks(
+    fresh_sync_status: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Phase 152 (task 04): the status carries the in-flight file's
+    VISION-image position — the pairs fire through the runner's OWN
+    ``_image_hook`` closure (task 03's wiring — the importer's
+    ``on_image_progress`` span's exact path), and the position CLIMBS
+    between ticks (3/15 → 15/15 — the PDF's vision span moving, the
+    owner's stuck label, now moving) while ``current_file`` and the
+    chunk position (14/38, a concurrent embed batch) keep their own
+    values — the THREE positions are independent sub-positions of the
+    file; the prelude reports 0/0, and the success terminal clears
+    BOTH sub-positions (image + chunk) to 0/0 while the FILE counts
+    keep their final values (the convention split)."""
+    clone = _GatedClone(threading.Event(), threading.Event())
+    image_release = threading.Event()
+    fake_import = _GatedImport(
+        ImportSummary(files=3, added=1, updated=1, unchanged=1),
+        threading.Event(),
+        threading.Event(),
+        chunk_pairs=[(14, 38)],  # a concurrent embed batch (independent)
+        image_pairs=[(3, 15), (15, 15)],  # two describe calls of a 15-image span
+        image_release=image_release,
+    )
+    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)
+
+    thread, errors = _start_run()
+    try:
+        assert clone.started.wait(5.0), "the run never reached the clone phase"
+        # Prelude: every position 0/0 (no file, no batch, no image
+        # yet — alongside the bare-label pins of the phase-64 test).
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["current_file"] is None
+        assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+        assert s["images_done"] == 0 and s["images_total"] == 0
+        clone.release.set()
+        assert fake_import.started.wait(5.0), "the run never reached the import"
+        # Import phase: the file is live (the progress hook's call),
+        # the chunk position 14/38 (the embed batch), and the image
+        # position 3/15 (the vision span) — the THREE positions are
+        # independent sub-positions of the same file.
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["phase"] == "import"
+        assert s["current_file"] == "repo/notes/deep.md"
+        assert s["files_done"] == 1 and s["files_total"] == 3
+        assert s["current_chunk"] == 14 and s["chunks_total"] == 38
+        assert s["images_done"] == 3 and s["images_total"] == 15
+        # Release the main park: the import fires image pair two
+        # (15/15 — the span's last describe call) and parks on
+        # image_release.
+        fake_import.release.set()
+        _poll_until(
+            lambda: sync_api.sync_status()["images_done"] == 15,
+            what="the second image pair on the status",
+        )
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["phase"] == "import"
+        # The image position CLIMBED (3 → 15 — monotonic, no
+        # backwards jump, never above the total) ... the FILE and
+        # chunk positions keep their own values (independent).
+        assert s["images_done"] == 15 and s["images_total"] == 15
+        assert s["current_file"] == "repo/notes/deep.md"
+        assert s["files_done"] == 1 and s["files_total"] == 3
+        assert s["current_chunk"] == 14 and s["chunks_total"] == 38
+        image_release.set()
+    finally:
+        clone.release.set()
+        fake_import.release.set()
+        image_release.set()
+        thread.join(10.0)
+    assert not thread.is_alive()
+    assert errors == [], f"the runner raised: {errors!r}"
+    # Wiring: every pair fired through the runner's OWN closure (the
+    # status reads above came through it; a wiring regression to
+    # None would have skipped the firing branch and timed out).
+    assert fake_import.chunk_hook_calls == [(14, 38)]
+    assert fake_import.image_hook_calls == [(3, 15), (15, 15)]
+    # Success terminal: BOTH sub-positions clear (0/0) ...
+    s = sync_api.sync_status()
+    assert s["state"] == "success"
+    assert s["phase"] is None
+    assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+    assert s["images_done"] == 0 and s["images_total"] == 0
+    # ... while the FILE counts keep their final values (the
+    # convention split, pinned alongside the phase-64 terminal pin).
+    assert s["current_file"] is None
+    assert s["files_done"] == 1 and s["files_total"] == 3
+
+
+class _TwoFileImport:
+    """The phase-152 (task 01 + task 04) fake: fires the runner's
+    OWN progress hook for file A, then file A's image position through
+    the runner's OWN ``on_image_progress`` closure (task 03's wiring —
+    what the importer's span would do), parks (the test reads the
+    live position on the status), fires the progress hook again for
+    file B (the runner's ``_hook`` under test must reset the image
+    position to 0/0 — the pairing rule, extended), parks again, and
+    completes."""
+
+    def __init__(self, summary: ImportSummary) -> None:
+        self.summary = summary
+        self.parked_a = threading.Event()  # parked after file A's progress
+        self.release_a = threading.Event()  # test releases → file B's progress
+        self.parked_b = threading.Event()  # parked after file B's progress
+        self.release_b = threading.Event()
+        self.hook_calls: list[tuple[str, str, int, int]] = []
+        self.image_hook_calls: list[tuple[int, int]] = []  # phase 152 (task 04)
+
+    async def __call__(
+        self,
+        sources: list[Path],
+        llm: object,
+        *,
+        prune: bool = False,
+        limit: int | None = None,
+        session: object = None,
+        progress: Callable[[str, str, int, int], None] | None = None,
+        on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
+        on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
+        ignore_by_root: dict[str, list[str]] | None = None,  # phase 89
+        include_hidden_by_root: dict[str, bool] | None = None,  # phase 105
+        doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,  # phase 106
+    ) -> ImportSummary:
+        if progress is not None:
+            progress("repo", "docs/a.pdf", 1, 2)
+            self.hook_calls.append(("repo", "docs/a.pdf", 1, 2))
+        # Phase 152 (task 04): file A's vision span is in flight —
+        # its image position (3/15) fires through the runner's OWN
+        # on_image_progress closure (task 03's wiring), then the fake
+        # parks.
+        if on_image_progress is not None:
+            on_image_progress(3, 15)
+            self.image_hook_calls.append((3, 15))
+        self.parked_a.set()
+        await asyncio.to_thread(self.release_a.wait)
+        if progress is not None:
+            # The SECOND progress call names file B — the closure under
+            # test must reset the image position the closure fired
+            # while file A was in flight (a file with no images must
+            # not inherit the previous PDF's position).
+            progress("repo", "docs/b.md", 2, 2)
+            self.hook_calls.append(("repo", "docs/b.md", 2, 2))
+        self.parked_b.set()
+        await asyncio.to_thread(self.release_b.wait)
+        return self.summary
+
+
+def test_image_position_resets_per_file(
+    fresh_sync_status: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Phase 152 (task 01 + task 04): the pairing rule, extended — a
+    ``progress`` call names the file AND clears the image position;
+    every image position until the next ``progress`` call belongs to
+    it. File A's live position (3/15, fired through the runner's OWN
+    ``_image_hook`` closure while parked on file A — task 04: the
+    phase-64/129 idiom, the importer's image hook's exact path) is
+    0/0 the moment the runner's own ``_hook`` names file B: a file
+    with no images must not inherit the previous PDF's position
+    (unlike the chunk position, not every file embeds chunks). The
+    success terminal keeps the position at 0/0 and the FILE counts
+    keep their final values (the convention split)."""
+    clone = _GatedClone(threading.Event(), threading.Event())
+    fake_import = _TwoFileImport(ImportSummary(files=2, added=1, updated=1))
+    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)
+
+    thread, errors = _start_run()
+    try:
+        assert clone.started.wait(5.0), "the run never reached the clone phase"
+        clone.release.set()
+        assert fake_import.parked_a.wait(5.0), "the run never parked on file A"
+        # File A is named by the runner's own _hook (a wiring
+        # regression to None would leave current_file null here).
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["phase"] == "import"
+        assert s["current_file"] == "repo/docs/a.pdf"
+        assert s["files_done"] == 1 and s["files_total"] == 2
+        # File A's vision span is in flight: its image position is
+        # live on the status — fired through the runner's OWN
+        # _image_hook closure (the fake fired it before parking;
+        # a wiring regression to None would leave it 0/0 here).
+        assert s["images_done"] == 3 and s["images_total"] == 15
+        # Release file A: the runner's _hook names file B — the image
+        # position must reset to 0/0 in the same call (the pairing
+        # rule, extended; the closure under test, driven through the
+        # fake's progress kwarg).
+        fake_import.release_a.set()
+        assert fake_import.parked_b.wait(5.0), "the run never parked on file B"
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["phase"] == "import"
+        assert s["current_file"] == "repo/docs/b.md"
+        assert s["files_done"] == 2 and s["files_total"] == 2
+        # RESET per file: file B (no images) does not inherit file
+        # A's 3/15.
+        assert s["images_done"] == 0 and s["images_total"] == 0
+        fake_import.release_b.set()
+    finally:
+        clone.release.set()
+        fake_import.release_a.set()
+        fake_import.release_b.set()
+        thread.join(10.0)
+    assert not thread.is_alive()
+    assert errors == [], f"the runner raised: {errors!r}"
+    # Wiring: BOTH progress calls went through the runner's own
+    # closure (the reset above was its doing), and the image pair
+    # went through the runner's own _image_hook closure.
+    assert fake_import.hook_calls == [
+        ("repo", "docs/a.pdf", 1, 2),
+        ("repo", "docs/b.md", 2, 2),
+    ]
+    assert fake_import.image_hook_calls == [(3, 15)]
+    # Success terminal: the image position stays 0/0 (cleared in the
+    # terminal too — every terminal clears it) and the FILE counts
+    # keep their final values (the convention split).
+    s = sync_api.sync_status()
+    assert s["state"] == "success"
+    assert s["phase"] is None
+    assert s["current_file"] is None
+    assert s["files_done"] == 2 and s["files_total"] == 2
+    assert s["images_done"] == 0 and s["images_total"] == 0
+
+
+def test_image_position_clears_at_phase_transitions_and_terminal(
+    fresh_sync_status: None,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """Phase 152 (task 01 + task 04): the image position is CLEARED
+    (0/0) at every point the import phase ends. This run's summary
+    changes the KB, so BOTH post-import transitions run and are
+    pinned: the import→``"overview"`` transition and the
+    import→``"summaries"`` transition (the changed-KB branch) each
+    clear the live position (3/15, fired mid-import through the
+    runner's OWN ``_image_hook`` closure — task 04: the
+    phase-64/129 idiom, the importer's span's exact path) to 0/0,
+    and the success terminal keeps it 0/0 — through the whole span
+    the FILE counts keep their final values (the convention split:
+    the file counts survive for the result line's context, the image
+    position is a sub-position of a file that's done)."""
+    clone = _GatedClone(threading.Event(), threading.Event())
+    import_release = threading.Event()
+    overview_started = threading.Event()
+    overview_release = threading.Event()
+    folders_started = threading.Event()
+    folders_release = threading.Event()
+    fake_import = _GatedImport(
+        ImportSummary(files=3, added=1, updated=1, unchanged=1),
+        threading.Event(),
+        import_release,
+        image_pairs=[(3, 15)],  # phase 152 (task 04): the live pair to clear
+    )
+    _patch_sync_seams(monkeypatch, tmp_path, fake_import, clone)
+
+    # Park the post-import steps (re-monkeypatched over the seams
+    # helper's no-ops): the test reads the status DURING the
+    # "overview" phase and DURING the "summaries" phase — the two
+    # import-phase-ending transitions the task pins (the same shape
+    # as the phase-129 chunk-position transition test above).
+    async def gated_overview(llm: object, session: object = None) -> bool:
+        overview_started.set()
+        await asyncio.to_thread(overview_release.wait)
+        return True
+
+    async def gated_folders(
+        db: object,
+        llm: object,
+        *,
+        skip: bool = False,
+        only_missing: bool = False,
+        changed_docs: Sequence[tuple[str, str]] | None = None,  # phase 127
+        on_progress: Callable[[int, int, str, str], None] | None = None,
+    ) -> dict[str, int]:
+        folders_started.set()
+        await asyncio.to_thread(folders_release.wait)
+        return {"generated": 0, "failed": 0, "pruned": 0}
+
+    monkeypatch.setattr(sync_api, "regenerate_overview", gated_overview)
+    monkeypatch.setattr(sync_api, "generate_folder_summaries", gated_folders)
+
+    thread, errors = _start_run()
+    try:
+        assert clone.started.wait(5.0), "the run never reached the clone phase"
+        clone.release.set()
+        assert fake_import.started.wait(5.0), "the run never reached the import"
+        # Mid-import: the file is live; its image position went live
+        # too — fired through the runner's OWN _image_hook closure
+        # (image_pairs[0] — the importer's span would do the same).
+        s = sync_api.sync_status()
+        assert s["phase"] == "import"
+        assert s["images_done"] == 3 and s["images_total"] == 15
+        # Release the import: it returns (added+updated > 0) → the
+        # phase machine moves to "overview" and the position clears.
+        import_release.set()
+        assert overview_started.wait(5.0), "the run never reached the overview"
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["phase"] == "overview"
+        # CLEARED at the import→overview transition (0/0) ... the file
+        # counts KEEP their final values (the convention split).
+        assert s["images_done"] == 0 and s["images_total"] == 0
+        assert s["files_done"] == 1 and s["files_total"] == 3
+        overview_release.set()
+        # ... and the changed-KB branch moves to "summaries" (the
+        # second import-phase-ending transition — cleared there too,
+        # idempotently) before parking.
+        assert folders_started.wait(5.0), "the run never reached the summaries"
+        s = sync_api.sync_status()
+        assert s["state"] == "running"
+        assert s["phase"] == "summaries"
+        assert s["images_done"] == 0 and s["images_total"] == 0
+        assert s["files_done"] == 1 and s["files_total"] == 3
+        folders_release.set()
+    finally:
+        clone.release.set()
+        import_release.set()
+        overview_release.set()
+        folders_release.set()
+        thread.join(10.0)
+    assert not thread.is_alive()
+    assert errors == [], f"the runner raised: {errors!r}"
+    assert fake_import.image_hook_calls == [(3, 15)]
+    # Success terminal: still 0/0 (the terminal clears too — same
+    # spots as current_file), file counts kept.
+    s = sync_api.sync_status()
+    assert s["state"] == "success"
+    assert s["phase"] is None
+    assert s["current_file"] is None
+    assert s["images_done"] == 0 and s["images_total"] == 0
+    assert s["files_done"] == 1 and s["files_total"] == 3
+
+
 # ---------- phase 134 (task 01): the cooperative cancel (5th state) ----------
 #
 # The cancel runs are driven through the REAL ``start_sync`` on a
@@ -1629,11 +2169,17 @@ def test_cancel_endpoint_stops_the_in_flight_run(
         ImportSummary(files=3, added=1, updated=1, unchanged=1),
         threading.Event(),
         threading.Event(),
+        image_pairs=[(3, 15)],  # phase 152 (task 04): the live pair before the cancel
     )
     _patch_sync_seams(monkeypatch, tmp_path, fake_import, _fast_clone)
 
     def act(loop: asyncio.AbstractEventLoop) -> dict[str, str]:
         assert fake_import.started.wait(5.0), "the run never reached the import"
+        # Phase 152 (task 04): the live image position in flight when
+        # the cancel converts was fired through the runner's OWN
+        # _image_hook closure (image_pairs[0] — the fake fired it
+        # before parking, like the importer's span would); the
+        # cancelled terminal must clear it.
         return loop.run_until_complete(sync_api.cancel_sync())
 
     try:
@@ -1665,6 +2211,11 @@ def test_cancel_endpoint_stops_the_in_flight_run(
     assert s["files_done"] == 1 and s["files_total"] == 3  # final counts kept
     assert s["summaries_done"] == 0 and s["summaries_total"] == 0
     assert s["current_chunk"] == 0 and s["chunks_total"] == 0
+    # Phase 152 (task 04): the image position clears in the CANCELLED
+    # terminal too — every terminal clears it (the live 3/15 fired
+    # through the closure in flight is gone; the FILE counts above
+    # keep their final values).
+    assert s["images_done"] == 0 and s["images_total"] == 0
     # PLAN §9: the ample-logging line for the converted cancel.
     assert any(
         r.getMessage()
@@ -1879,6 +2430,7 @@ def test_event_set_during_import_cancels_before_the_lite_spans(
             session: object = None,
             progress: Callable[[str, str, int, int], None] | None = None,
             on_chunk_progress: Callable[[int, int], None] | None = None,  # phase 129
+            on_image_progress: Callable[[int, int], None] | None = None,  # phase 152
             ignore_by_root: dict[str, list[str]] | None = None,  # phase 89
             include_hidden_by_root: dict[str, bool] | None = None,  # phase 105
             doc_dates_by_root: dict[str, dict[str, datetime]] | None = None,  # phase 106