diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 54b077a1..605a7530 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -621,7 +621,9 @@ The tool implements a SQLite-based caching system for performance:
 - **Concurrency**: WAL mode, one writer at a time. Every write goes through
   `cache.write_transaction`; in the parallel pass, project workers take
   turns through a fair cross-process queue (`write_queue.py`), so none gives
-  up on a sibling. Against other processes connections wait 30 s
+  up on a sibling. Workers write only rows: an existing search index is
+  caught up once at the end of the pass, by the parent, not inside each
+  worker's write transaction. Against other processes connections wait 30 s
   (`CLAUDE_CODE_LOG_CACHE_TIMEOUT` overrides, in seconds), and a project
   locked out is retried after the pool rather than failed. A pass compacts
   an oversized WAL before it starts and truncates the WAL when it ends. A
diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py
index aa10595c..7ddbb689 100644
--- a/claude_code_log/cache.py
+++ b/claude_code_log/cache.py
@@ -1030,6 +1030,54 @@ def checkpoint_wal(
             conn.close()
 
 
+def catch_up_search_index(db_path: Path) -> int:
+    """Bring an existing search index up to date after a hierarchy pass.
+
+    The pass's entry writes leave the index alone (see
+    `search.index_maintenance_deferred`), so this is where the files they
+    rewrote get re-indexed: once, serially, on this thread's lease when
+    there is one, after the pool has exited — so the tokenising never
+    holds the write lock against a sibling. Returns the number of files
+    refreshed; 0 when no index exists or maintenance is switched off
+    (`CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0`).
+
+    The index is derived data that the next pass heals (its entries stay
+    behind their cache rows until re-extracted), so no error here is
+    worth failing the pass over: the pages are written already. A cache
+    busy with another process's writer, a full disk, a damaged index and
+    an undecodable blob are all reported with the same warning and
+    return 0 — unlike an entry write, where the same error fails that
+    one project's save and nothing else.
+    """
+    from .search import auto_index_enabled, catch_up_index
+
+    if not auto_index_enabled() or not db_path.exists():
+        return 0
+    leased = _thread_leases().get((str(db_path), False))
+    try:
+        conn = leased or _open_connection(
+            db_path,
+            read_only=False,
+            configure=functools.partial(_configure_connection, read_only=False),
+        )
+    except sqlite3.Error:
+        return 0
+    try:
+        return catch_up_index(conn)
+    # ValueError covers what `search.decode_entry` raises past zlib: a blob
+    # that decompresses but is not UTF-8 (UnicodeDecodeError) or not JSON
+    # (JSONDecodeError) — both subclasses, neither a zlib.error.
+    except (sqlite3.Error, zlib.error, ValueError) as exc:
+        print(
+            "Warning: the search index was not brought up to date "
+            f"({exc}); the next run or `serve` start will do it."
+        )
+        return 0
+    finally:
+        if leased is None:
+            conn.close()
+
+
 def trim_oversized_wal(
     db_path: Path, *, announce: Callable[[str], None] = print
 ) -> Optional[int]:
@@ -1711,9 +1759,16 @@ def save_cached_entries(
             # search pay essentially nothing — and those who do never meet a
             # stale index, because the file's old rows go out with its old
             # messages rather than waiting for the next server start.
-            from .search import auto_index_enabled, reindex_files
+            # Inside a hierarchy pass the tokenising would run under the
+            # write lock in every worker, so the pass defers it to one
+            # catch-up at its end (`catch_up_search_index`).
+            from .search import (
+                auto_index_enabled,
+                index_maintenance_deferred,
+                reindex_files,
+            )
 
-            if auto_index_enabled():
+            if auto_index_enabled() and not index_maintenance_deferred():
                 reindex_files(conn, [file_id], commit=False)
 
             self._mark_entries_written(conn)
@@ -1849,10 +1904,15 @@ def _append_under_lock(
         # The index still refreshes the whole file: `reindex_files`
         # deletes and re-adds its rows, and an append-only variant
         # would need its own correctness argument. Only users who have
-        # built an index pay it, exactly as before.
-        from .search import auto_index_enabled, reindex_files
+        # built an index pay it, exactly as before — and a hierarchy
+        # pass defers it to its end, as `save_cached_entries` does.
+        from .search import (
+            auto_index_enabled,
+            index_maintenance_deferred,
+            reindex_files,
+        )
 
-        if auto_index_enabled():
+        if auto_index_enabled() and not index_maintenance_deferred():
             reindex_files(conn, [file_id], commit=False)
 
         self._mark_entries_written(conn)
diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py
index f09c7d5f..67ff583c 100644
--- a/claude_code_log/converter.py
+++ b/claude_code_log/converter.py
@@ -59,6 +59,7 @@
 from .cache import (
     CacheManager,
     SessionCacheData,
+    catch_up_search_index,
     checkpoint_wal,
     connection_lease,
     get_all_cached_projects,
@@ -6507,6 +6508,16 @@ def _plan_project(
     )
 
 
+def _init_project_worker(turns: Optional[WriteTurns]) -> None:
+    """The project pool's initializer: the pass-wide write queue, and no
+    search-index work under the write lock (the parent catches up once
+    the pool has exited — `catch_up_search_index`)."""
+    from .search import defer_index_maintenance
+
+    install_write_turns(turns)
+    defer_index_maintenance()
+
+
 def _convert_project_worker(
     worker_args: Dict[str, Any],
 ) -> "tuple[str, float, Optional[str], bool]":
@@ -6573,6 +6584,8 @@ def process_projects_hierarchy(
     it, so a caller can tell a complete pass from one that lost projects
     (the CLI exits non-zero on the latter).
     """
+    from .search import deferred_index_maintenance
+
     lease = (
         connection_lease(get_cache_db_path(projects_path))
         if use_cache
@@ -6583,33 +6596,45 @@ def process_projects_hierarchy(
             # Inside the lease, so the connection that recovers the log
             # is the one the pass then keeps.
             trim_oversized_wal(get_cache_db_path(projects_path))
-        result = _process_projects_hierarchy(
-            projects_path=projects_path,
-            from_date=from_date,
-            to_date=to_date,
-            use_cache=use_cache,
-            generate_individual_sessions=generate_individual_sessions,
-            output_format=output_format,
-            image_export_mode=image_export_mode,
-            silent=silent,
-            page_size=page_size,
-            depth=depth,
-            compact=compact,
-            output_dir=output_dir,
-            expand_paths=expand_paths,
-            filter_path=filter_path,
-            write_combined=write_combined,
-            no_timestamps=no_timestamps,
-            no_recaps=no_recaps,
-            theme=theme,
-            jobs=jobs,
-            entry_store=entry_store,
-            failed_projects=failed_projects,
-        )
+        # No entry write of this pass re-indexes its file for search —
+        # not in the pool (its initializer defers too) and not inline
+        # here: that is tokenising under the database's single write
+        # lock. The pass catches the index up once, below.
+        with deferred_index_maintenance():
+            result = _process_projects_hierarchy(
+                projects_path=projects_path,
+                from_date=from_date,
+                to_date=to_date,
+                use_cache=use_cache,
+                generate_individual_sessions=generate_individual_sessions,
+                output_format=output_format,
+                image_export_mode=image_export_mode,
+                silent=silent,
+                page_size=page_size,
+                depth=depth,
+                compact=compact,
+                output_dir=output_dir,
+                expand_paths=expand_paths,
+                filter_path=filter_path,
+                write_combined=write_combined,
+                no_timestamps=no_timestamps,
+                no_recaps=no_recaps,
+                theme=theme,
+                jobs=jobs,
+                entry_store=entry_store,
+                failed_projects=failed_projects,
+            )
         if use_cache:
-            # The pass's workers have all exited, so this is the moment the
-            # WAL can be reset — see `checkpoint_wal` (issue #332).
-            checkpoint_wal(get_cache_db_path(projects_path))
+            # The pass's workers have all exited: the files they rewrote
+            # are re-indexed for search now, serially, with no sibling
+            # waiting on the lock — and then the WAL can be reset, see
+            # `checkpoint_wal` (issue #332). The catch-up reports its own
+            # failures; the `finally` is for whatever it did not foresee,
+            # so the checkpoint is never the casualty of the index.
+            try:
+                catch_up_search_index(get_cache_db_path(projects_path))
+            finally:
+                checkpoint_wal(get_cache_db_path(projects_path))
         return result
 
 
@@ -7069,7 +7094,7 @@ def _convert_plan_inline(
             with ProcessPoolExecutor(
                 max_workers=resolved_jobs,
                 mp_context=ctx,
-                initializer=install_write_turns,
+                initializer=_init_project_worker,
                 initargs=(WriteTurns(ctx),),
             ) as pool:
                 future_to_plan = {
diff --git a/claude_code_log/search.py b/claude_code_log/search.py
index 7abb83b7..9fd29961 100644
--- a/claude_code_log/search.py
+++ b/claude_code_log/search.py
@@ -38,6 +38,7 @@
 import re
 import sqlite3
 import zlib
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any, Callable, Iterable, Iterator, Optional, Sequence, cast
@@ -46,6 +47,7 @@
 # cycle. The name is prefixed to keep it visually separate from the index
 # internals defined below — it answers a cache-file question, not an FTS one.
 from .cache import is_corrupt_database_error as _is_corrupt_database_error
+from .cache import write_transaction
 
 # ---------------------------------------------------------------------------
 # Field groups
@@ -730,9 +732,11 @@ def ensure_index(
     )
 
     for done, file_id in enumerate(todo, start=1):
-        if file_id in indexed_files:
-            # Re-indexing a changed file: clear its old rows first.
-            _delete_file_rows(conn, file_id)
+        # Clear the file's old rows first — unconditionally, not only when
+        # the snapshot above knew the file: another process may have
+        # indexed it since, and a contentless FTS5 table accepts the
+        # duplicate rowids without a word. A no-op when there are none.
+        _delete_file_rows(conn, file_id)
         _index_file(conn, file_id, fields, insert_sql)
         conn.execute(
             f"INSERT INTO {FILES_TABLE}"
@@ -779,6 +783,121 @@ def auto_index_enabled() -> bool:
     return value.strip().lower() not in ("0", "false", "no", "off")
 
 
+# Depth of `deferred_index_maintenance` scopes in this process. A pool
+# worker's initializer sets it for the worker's whole life.
+_deferred_depth = 0
+
+
+def index_maintenance_deferred() -> bool:
+    """Whether entry writes should leave the index to a later `catch_up_index`.
+
+    Inside a hierarchy pass the answer is yes, in the parent and in every
+    project worker: re-indexing a file means decompressing and tokenising
+    its messages, and done inside the entry write it runs under the cache
+    database's single write lock — the longest thing any writer did while
+    holding it (3.4 s of an 11.2 s save on the largest session measured),
+    and in a pool it lengthened every sibling's wait. The pass catches up
+    once at its end instead, on the parent's connection, with the pool
+    gone. Outside a pass (a single-file conversion, the TUI) nothing is
+    deferred and the entry write maintains the index as before.
+    """
+    return _deferred_depth > 0
+
+
+def defer_index_maintenance() -> None:
+    """Defer index maintenance for the rest of this process's life.
+
+    For a pool worker's initializer: the worker only ever runs pass
+    conversions, and the parent's catch-up covers them all.
+    """
+    global _deferred_depth
+    _deferred_depth += 1
+
+
+@contextmanager
+def deferred_index_maintenance() -> Iterator[None]:
+    """Defer index maintenance for the scope (nests; the outermost restores)."""
+    global _deferred_depth
+    _deferred_depth += 1
+    try:
+        yield
+    finally:
+        _deferred_depth -= 1
+
+
+def catch_up_index(conn: sqlite3.Connection) -> int:
+    """Bring an existing index up to date with `cached_files`; return the
+    number of files touched (re-indexed, added or removed).
+
+    The end-of-pass counterpart of `reindex_files`: every file the pass
+    rewrote has a new `cached_files.cached_mtime`, so the same comparison
+    `ensure_index` makes finds them, and vanished files take their rows
+    with them. Builds nothing when no index exists, and leaves an index
+    whose table cannot DELETE to the next `ensure_index`, which rebuilds
+    it (the same rule `reindex_files` follows).
+
+    One transaction per file, opened with the cache's `write_transaction`
+    so each hold on the write lock is one file's tokenising and an
+    interrupted catch-up resumes where it stopped.
+    """
+    if not _index_exists(conn) or not _index_delete_supported(conn):
+        return 0
+    fields = indexed_fields(conn)
+    if not fields:
+        return 0
+
+    indexed_files = {
+        int(row[0]): float(row[1] or 0.0)
+        for row in conn.execute(f"SELECT file_id, cached_mtime FROM {FILES_TABLE}")
+    }
+    current_files = {
+        int(row[0]): (int(row[1] or 0), float(row[2] or 0.0))
+        for row in conn.execute(
+            "SELECT id, message_count, cached_mtime FROM cached_files"
+        )
+    }
+    gone = set(indexed_files) - set(current_files)
+    todo = [
+        file_id
+        for file_id, (_, mtime) in current_files.items()
+        if file_id not in indexed_files or abs(indexed_files[file_id] - mtime) >= 1e-6
+    ]
+    if not gone and not todo:
+        return 0
+
+    placeholders = ",".join("?" * (len(fields) + 1))
+    insert_sql = (
+        f"INSERT INTO {FTS_TABLE}(rowid, {', '.join(fields)}) VALUES({placeholders})"
+    )
+    now = datetime.now(timezone.utc).isoformat()
+    if gone:
+        with write_transaction(conn):
+            for file_id in gone:
+                _delete_file_rows(conn, file_id)
+                conn.execute(f"DELETE FROM {FILES_TABLE} WHERE file_id = ?", (file_id,))
+    for file_id in todo:
+        with write_transaction(conn):
+            # Inside the transaction and unconditional: the snapshot above
+            # is stale by the time the lock is held, and a file another
+            # process indexed meanwhile would otherwise get its rowids
+            # inserted twice (which a contentless table does not refuse).
+            _delete_file_rows(conn, file_id)
+            _index_file(conn, file_id, fields, insert_sql)
+            message_count, mtime = current_files[file_id]
+            conn.execute(
+                f"INSERT INTO {FILES_TABLE}"
+                "(file_id, indexed_at, message_count, cached_mtime) "
+                "VALUES(?, ?, ?, ?) ON CONFLICT(file_id) DO UPDATE SET "
+                "indexed_at = excluded.indexed_at, "
+                "message_count = excluded.message_count, "
+                "cached_mtime = excluded.cached_mtime",
+                (file_id, now, message_count, mtime),
+            )
+    with write_transaction(conn):
+        _refresh_indexed_count(conn)
+    return len(todo) + len(gone)
+
+
 def reindex_files(
     conn: sqlite3.Connection,
     file_ids: Sequence[int],
@@ -793,7 +912,10 @@ def reindex_files(
     run `serve` pay one `sqlite_master` lookup and nothing else.
 
     Cheap enough for that path: ~180 ms for the largest file in an 8 GB
-    archive, and only files the cache actually rewrote are touched.
+    archive, and only files the cache actually rewrote are touched. Not
+    for a hierarchy pass, though: there the entry write holds the
+    database's single write lock, so the pass defers to `catch_up_index`
+    (see `index_maintenance_deferred`).
 
     Pass `commit=False` to join the caller's transaction rather than
     committing on its own.
diff --git a/dev-docs/application_model.md b/dev-docs/application_model.md
index 183a5a47..0e4b9c58 100644
--- a/dev-docs/application_model.md
+++ b/dev-docs/application_model.md
@@ -186,6 +186,35 @@ Uncontended, the queue plus `BEGIN IMMEDIATE` add ~10 µs to the smallest
 write (`update_html_cache`, 46 → 56 µs) and nothing measurable to a
 session save; whole passes are unchanged at `--jobs 1` and within ~4%
 on a pooled incremental pass, the price of strict turn-taking.
+The queue makes waiting safe; it does not shorten the holds, and the
+longest thing a worker did while holding the lock was not the row
+insert but maintaining the search index: `save_cached_entries` called
+`search.reindex_files` inside its write transaction, so once an index
+exists every rewritten file was decompressed and tokenised under the
+lock (3.4 s of an 11.2 s save on the largest session measured, 122 MB
+/ 46k entries), in every worker, with the rest of the pool queued
+behind it. So a hierarchy pass **defers index maintenance**
+(`search.deferred_index_maintenance` in the parent,
+`search.defer_index_maintenance` in the pool initializer
+`_init_project_worker`): the entry writes leave the index alone, and
+after `_process_projects_hierarchy` returns — pool exited, hold-backs
+and retries done — `cache.catch_up_search_index` re-indexes the files
+whose `cached_files.cached_mtime` moved, one short transaction per
+file on the pass's leased connection, before the WAL checkpoint. What
+this buys is shorter holds, not a shorter pass: on 32 large projects
+(9 GB, 3,201 transcripts all touched, index present, `--jobs 16`) the
+parallel phase went from 555 s to 230 s and the longest
+`save_cached_entries` from 44.8 s to 15.2 s (p90 6.8 s → 2.8 s), but
+the serial catch-up then took ~305 s, so the whole run was 555 s
+against 536 s. The tokenising is the same work wherever it lands; in
+the workers it was paid under the lock, and a hold that long is what
+pushes a writer in *another* process (the TUI, a second `serve`) past
+its 30 s budget. On the daily shape — a handful of files changed —
+the catch-up is seconds. Parallelising it (extract in workers, insert
+in the parent) is the open lever if that worst case ever matters. A
+single file or directory conversion (`convert_jsonl_to` with no pass
+around it) still re-indexes inside the write, where nothing competes
+for the lock. `CLAUDE_CODE_LOG_SEARCH_AUTO_INDEX=0` switches both off.
 `serve` takes the same `--jobs` as the main command. A pool-level failure (e.g. a library caller without the
 `if __name__ == "__main__"` guard `spawn` needs) degrades to inline
 sequential processing with a warning rather than aborting.
diff --git a/docs/archive-search.md b/docs/archive-search.md
index 1c4fbd62..ce4e304b 100644
--- a/docs/archive-search.md
+++ b/docs/archive-search.md
@@ -160,6 +160,12 @@ The index is derived data: deleting the cache rebuilds both.
 Once an index exists, ordinary `claude-code-log` runs keep it up to date —
 you don't have to restart the server to search new conversations.
 
+A run over all projects re-indexes the transcripts it rewrote at its end,
+once, after the per-project workers have finished: done inside each
+worker's cache writes it held the cache's single write lock while
+tokenising, and the other workers waited. Converting a single file or
+directory re-indexes as it writes, as before.
+
 That costs about 25% extra on a conversion in which *every* transcript
 changed (3.3 s → 4.1 s over 7,516 messages); a normal run touches a handful
 of files and the difference isn't noticeable. Nothing is added for people
diff --git a/test/test_search.py b/test/test_search.py
index f89e35ff..1cb15e32 100644
--- a/test/test_search.py
+++ b/test/test_search.py
@@ -952,6 +952,318 @@ def test_conversion_does_not_create_an_index_for_non_users(tmp_path: Path) -> No
         conn.close()
 
 
+# ---------------------------------------------------------------------------
+# Maintenance deferred to the end of a hierarchy pass
+# ---------------------------------------------------------------------------
+
+
+def test_catch_up_index_refreshes_new_changed_and_removed_files(
+    cache: sqlite3.Connection,
+) -> None:
+    """The end-of-pass catch-up does what `ensure_index` does for an index
+    that already exists: new files in, rewritten files replaced, vanished
+    files out — and reports how many it touched."""
+    from claude_code_log.search import catch_up_index
+
+    ensure_index(cache)
+    assert catch_up_index(cache) == 0, "an unchanged cache has nothing to catch up"
+
+    # A rewritten file: new rows, re-stamped cached_mtime (what the cache does).
+    cache.execute("DELETE FROM messages WHERE file_id = 1")
+    _add_message(cache, 30, 1, 1, _assistant("now mentions mercurial"), uuid="uuid-30")
+    cache.execute("UPDATE cached_files SET cached_mtime = 12345.678 WHERE id = 1")
+    # A new file.
+    cache.execute(
+        "INSERT INTO cached_files(id, project_id, file_name, file_path, "
+        "source_mtime, cached_mtime, message_count) "
+        "VALUES(3, 1, 'f3.jsonl', '/tmp/f3.jsonl', 0, 1.0, 1)"
+    )
+    _add_message(cache, 20, 1, 3, _assistant("a brand new file"), uuid="uuid-20")
+    # A vanished file.
+    cache.execute("DELETE FROM messages WHERE file_id = 2")
+    cache.execute("DELETE FROM cached_files WHERE id = 2")
+    cache.commit()
+
+    assert catch_up_index(cache) == 3, "one rewritten, one new, one vanished"
+    assert count_matches(cache, "mercurial") == 1
+    assert count_matches(cache, "brand") == 1
+    assert count_matches(cache, "pydantic") == 0, "stale or vanished rows survived"
+    assert index_status(cache).indexed_messages == 2
+    assert catch_up_index(cache) == 0
+
+
+def test_catch_up_index_without_an_index_builds_nothing(
+    cache: sqlite3.Connection,
+) -> None:
+    """Catching up is maintenance of an index someone built; it never creates one."""
+    from claude_code_log.search import catch_up_index
+
+    assert catch_up_index(cache) == 0
+    row = cache.execute(
+        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?",
+        (FTS_TABLE,),
+    ).fetchone()
+    assert row is None
+
+
+def test_deferred_index_maintenance_is_scoped() -> None:
+    from claude_code_log.search import (
+        deferred_index_maintenance,
+        index_maintenance_deferred,
+    )
+
+    assert not index_maintenance_deferred()
+    with deferred_index_maintenance():
+        assert index_maintenance_deferred()
+        with deferred_index_maintenance():
+            assert index_maintenance_deferred()
+        assert index_maintenance_deferred(), "an inner scope ended the outer one"
+    assert not index_maintenance_deferred()
+
+
+def _report_deferred() -> bool:
+    from claude_code_log.search import index_maintenance_deferred
+
+    return index_maintenance_deferred()
+
+
+def test_project_pool_workers_defer_index_maintenance() -> None:
+    """A project worker never tokenises under the cache's write lock: the
+    pool's initializer defers the index work to the parent's catch-up."""
+    import multiprocessing
+    from concurrent.futures import ProcessPoolExecutor
+
+    from claude_code_log.converter import _init_project_worker  # pyright: ignore[reportPrivateUsage]
+    from claude_code_log.write_queue import WriteTurns
+
+    ctx = multiprocessing.get_context("spawn")
+    with ProcessPoolExecutor(
+        max_workers=1,
+        mp_context=ctx,
+        initializer=_init_project_worker,
+        initargs=(WriteTurns(ctx),),
+    ) as pool:
+        assert pool.submit(_report_deferred).result(timeout=60) is True
+
+
+def test_a_hierarchy_pass_maintains_the_index_after_the_pool_not_inside_the_writes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The guarantee of `test_a_normal_conversion_keeps_an_existing_index_current`
+    is kept by one catch-up at the end of the pass, not by every entry
+    write re-indexing its file inside the write transaction (where the
+    tokenising held the database's single write lock, in every worker).
+    With the catch-up stubbed out, the pass must leave the index stale."""
+    from claude_code_log import converter
+    from claude_code_log.cache import get_cache_db_path
+    from claude_code_log.converter import process_projects_hierarchy
+
+    projects = tmp_path / "projects"
+    project = projects / "-home-u-proj"
+    project.mkdir(parents=True)
+    jsonl = project / "session.jsonl"
+    jsonl.write_text(_entry_line("the first version mentions kangaroo"))
+    process_projects_hierarchy(projects, silent=True)
+
+    db_path = get_cache_db_path(projects)
+    conn = sqlite3.connect(db_path)
+    ensure_index(conn)
+    assert count_matches(conn, "kangaroo") == 1
+    conn.close()
+
+    calls: list[Path] = []
+
+    def record(path: Path) -> int:
+        calls.append(path)
+        return 0
+
+    monkeypatch.setattr(converter, "catch_up_search_index", record)
+
+    jsonl.write_text(_entry_line("the second version mentions platypus"))
+    bump_mtime(jsonl)
+    process_projects_hierarchy(projects, silent=True)
+
+    assert calls == [db_path], "the pass did not run one catch-up on its cache"
+    conn = sqlite3.connect(db_path)
+    try:
+        assert count_matches(conn, "platypus") == 0, (
+            "the entry write maintained the index itself"
+        )
+        assert count_matches(conn, "kangaroo") == 1
+    finally:
+        conn.close()
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        pytest.param(zlib.error("incorrect header check"), id="zlib"),
+        pytest.param(
+            sqlite3.OperationalError("database or disk is full"), id="disk-full"
+        ),
+        pytest.param(
+            sqlite3.DatabaseError("database disk image is malformed"), id="malformed"
+        ),
+    ],
+)
+def test_a_catch_up_error_does_not_fail_the_pass(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    error: Exception,
+) -> None:
+    """The index is derived data that the next pass heals, so an error
+    while catching it up is a warning: the pass returns, loses no project,
+    and still checkpoints the WAL."""
+    from claude_code_log import converter, search
+    from claude_code_log.cache import get_cache_db_path
+    from claude_code_log.converter import process_projects_hierarchy
+
+    projects = tmp_path / "projects"
+    project = projects / "-home-u-proj"
+    project.mkdir(parents=True)
+    jsonl = project / "session.jsonl"
+    jsonl.write_text(_entry_line("the first version mentions kangaroo"))
+    process_projects_hierarchy(projects, silent=True)
+    db_path = get_cache_db_path(projects)
+    conn = sqlite3.connect(db_path)
+    ensure_index(conn)
+    conn.close()
+
+    checkpoints: list[Path] = []
+    monkeypatch.setattr(
+        converter, "checkpoint_wal", lambda path: checkpoints.append(path)
+    )
+    real_catch_up = search.catch_up_index
+
+    def failing(conn: sqlite3.Connection) -> int:
+        raise error
+
+    monkeypatch.setattr(search, "catch_up_index", failing)
+    jsonl.write_text(_entry_line("the second version mentions platypus"))
+    bump_mtime(jsonl)
+    failed: list[str] = []
+    process_projects_hierarchy(projects, silent=True, failed_projects=failed)
+
+    assert failed == []
+    assert checkpoints == [db_path], "the WAL checkpoint was skipped"
+    assert "search index was not brought up to date" in capsys.readouterr().out
+
+    # Nothing changed on disk, but the file's index entry is behind its
+    # cache row, so the next pass's catch-up heals it.
+    monkeypatch.setattr(search, "catch_up_index", real_catch_up)
+    process_projects_hierarchy(projects, silent=True)
+    conn = sqlite3.connect(db_path)
+    try:
+        assert count_matches(conn, "platypus") == 1
+        assert count_matches(conn, "kangaroo") == 0
+    finally:
+        conn.close()
+
+
+@pytest.mark.parametrize(
+    "blob",
+    [
+        pytest.param(zlib.compress(b"\xff\xfe\x00 not utf-8"), id="not-utf8"),
+        pytest.param(zlib.compress(b"{not json"), id="not-json"),
+    ],
+)
+def test_an_undecodable_blob_in_the_catch_up_does_not_fail_the_pass(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    blob: bytes,
+) -> None:
+    """A blob that decompresses but is not UTF-8 JSON raises a ValueError
+    (UnicodeDecodeError, JSONDecodeError) from `decode_entry`, not a
+    zlib.error: the same warning, the pass returns and still checkpoints,
+    and the next pass that rewrites the file heals the index."""
+    from claude_code_log import converter
+    from claude_code_log.cache import get_cache_db_path
+    from claude_code_log.converter import process_projects_hierarchy
+
+    projects = tmp_path / "projects"
+    project = projects / "-home-u-proj"
+    project.mkdir(parents=True)
+    jsonl = project / "session.jsonl"
+    jsonl.write_text(_entry_line("the first version mentions kangaroo"))
+    process_projects_hierarchy(projects, silent=True)
+    db_path = get_cache_db_path(projects)
+    conn = sqlite3.connect(db_path)
+    ensure_index(conn)
+    # The file's cache row moved past its index entry (what an entry write
+    # does), and one of its blobs cannot be decoded.
+    conn.execute(
+        "UPDATE messages SET content = ? WHERE id = "
+        "(SELECT min(id) FROM messages WHERE file_id = 1)",
+        (blob,),
+    )
+    conn.execute("UPDATE cached_files SET cached_mtime = cached_mtime + 1 WHERE id = 1")
+    conn.commit()
+    conn.close()
+
+    checkpoints: list[Path] = []
+    monkeypatch.setattr(
+        converter, "checkpoint_wal", lambda path: checkpoints.append(path)
+    )
+    failed: list[str] = []
+    process_projects_hierarchy(projects, silent=True, failed_projects=failed)
+
+    assert failed == []
+    assert checkpoints == [db_path], "the WAL checkpoint was skipped"
+    assert "search index was not brought up to date" in capsys.readouterr().out
+
+    jsonl.write_text(_entry_line("the second version mentions platypus"))
+    bump_mtime(jsonl)
+    process_projects_hierarchy(projects, silent=True)
+    conn = sqlite3.connect(db_path)
+    try:
+        assert count_matches(conn, "platypus") == 1
+        assert count_matches(conn, "kangaroo") == 0
+    finally:
+        conn.close()
+
+
+def test_catch_up_index_tolerates_a_file_indexed_since_its_snapshot(
+    cache: sqlite3.Connection, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A contentless FTS5 table accepts a duplicate rowid without complaint,
+    so a file another process indexed between the catch-up's snapshot and
+    its transaction must have its rows cleared before being re-added, or
+    its tokens count twice."""
+    from contextlib import contextmanager
+
+    from claude_code_log import search
+
+    ensure_index(cache)
+    cache.execute(
+        "INSERT INTO cached_files(id, project_id, file_name, file_path, "
+        "source_mtime, cached_mtime, message_count) "
+        "VALUES(3, 1, 'f3.jsonl', '/tmp/f3.jsonl', 0, 1.0, 1)"
+    )
+    _add_message(cache, 20, 1, 3, _assistant("a brand new file"), uuid="uuid-20")
+
+    real_write_transaction = search.write_transaction
+    raced = False
+
+    @contextmanager
+    def racing(conn: sqlite3.Connection):
+        nonlocal raced
+        if not raced:
+            raced = True
+            # Another process indexes the same new file and commits, after
+            # our snapshot and before our transaction.
+            reindex_files(conn, [3])
+        with real_write_transaction(conn):
+            yield
+
+    monkeypatch.setattr(search, "write_transaction", racing)
+    search.catch_up_index(cache)
+
+    assert raced
+    assert count_matches(cache, "brand") == 1, "the file's tokens were indexed twice"
+
+
 def _entry_line(text: str) -> str:
     """One user message as a JSONL line."""
     return (