perf(daemon): bound catch-up chunk convergence to the chunk's own input

polylogue/daemon/cli.py

@@ -947,13 +947,21 @@ async def _periodic_convergence_check(
     sources: tuple[WatchSource, ...],
     *,
     catch_up_complete: asyncio.Event | None = None,
+    catch_up_active: Callable[[], bool] | None = None,
 ) -> None:
-    """Periodically retry recorded derived convergence debt."""
+    """Periodically retry recorded derived convergence debt.
+
+    The archive-wide exact FTS audit is whole-archive work; while the watcher
+    is inside a chunked catch-up it is skipped (the catch-up's last chunk
+    publishes readiness) instead of rescanning the growing archive between
+    chunks.
+    """
     db = _active_index_db_path()
     await _await_catch_up_gate(catch_up_complete, loop_name="convergence debt retry")
     while True:
         await _retry_convergence_debt_once(db)
-        await _run_periodic_fts_convergence_once(db)
+        if catch_up_active is None or not catch_up_active():
+            await _run_periodic_fts_convergence_once(db)
         await asyncio.sleep(_CONVERGENCE_DEBT_RETRY_INTERVAL_SECONDS)
 
 
@@ -3079,6 +3087,9 @@ async def _run_daemon_services_under_active_writer_lease(
         # generation before publishing either HTTP socket.  Otherwise an
         # immediate similarity request can recreate legacy WAL sidecars after
         # the lifecycle checkpoint and turn a clean restart into a failure.
+        # Filled once the watcher exists; maintenance loops consult it for
+        # catch-up activity without holding the watcher before creation.
+        watcher_holder: list[LiveWatcher] = []
         if not watcher_blocked:
             await _run_startup_embedding_lifecycle(write_coordinator, archive_root_path)
             if lifecycle_events_enabled:
@@ -3163,7 +3174,11 @@ async def _run_daemon_services_under_active_writer_lease(
             catch_up_complete_gate = asyncio.Event() if enable_watch else None
             periodic_loops = [
                 _periodic_raw_materialization_convergence(catch_up_complete=catch_up_complete_gate),
-                _periodic_convergence_check(sources, catch_up_complete=catch_up_complete_gate),
+                _periodic_convergence_check(
+                    sources,
+                    catch_up_complete=catch_up_complete_gate,
+                    catch_up_active=lambda: bool(watcher_holder and watcher_holder[0].catch_up_active),
+                ),
                 _periodic_wal_checkpoint(),
                 _periodic_fts_merge(),
                 _periodic_heartbeat(),
@@ -3223,6 +3238,7 @@ async def _run_daemon_services_under_active_writer_lease(
                         catch_up_event_emitter=emit_catch_up_cycle,
                         write_coordinator=write_coordinator,
                     )
+                    watcher_holder.append(watcher)
                     watcher_catch_up_complete = getattr(watcher, "catch_up_complete", None)
                     if catch_up_complete_gate is not None and watcher_catch_up_complete is not None:
                         maintenance_tasks.append(

polylogue/daemon/convergence.py

@@ -82,6 +82,12 @@ class ConvergenceStage:
     barrier_check_sessions: Callable[[Sequence[str]], set[str]] | None = None
     # Optional bounded, secret-safe operator status payload.
     status: Callable[[], Mapping[str, object]] | None = None
+    # The stage's work is a function of the whole archive, not of the batch's
+    # subjects (a graph rebuilt from every raw artifact, an exact archive-wide
+    # readiness audit). ``converge_batch(whole_archive=False)`` skips such
+    # stages so a catch-up chunk's cost stays bounded by its own input; the
+    # catch-up's final chunk runs them once for the whole backlog.
+    whole_archive: bool = False
 
 
 @dataclass(slots=True)
@@ -345,8 +351,19 @@ def _evict_converged_sessions(self, session_ids: Iterable[str]) -> None:
             if state is not None and state.converged:
                 del self._session_states[session_id]
 
-    def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState], dict[str, float]]:
-        """Converge a changed source batch with per-subject stage barriers."""
+    def converge_batch(
+        self, files: Iterable[Path], *, whole_archive: bool = True
+    ) -> tuple[dict[Path, FileState], dict[str, float]]:
+        """Converge a changed source batch with per-subject stage barriers.
+
+        ``whole_archive=False`` bounds the pass to the batch's own subjects:
+        stages declared ``whole_archive`` are recorded ``SKIPPED`` (converged,
+        no debt) because their staleness is re-derived from archive content by
+        the next whole-archive pass, never from this batch's outcome.
+
+        Stage ``check``/``check_many`` time is charged to ``<stage>.check`` in
+        the returned ledger so the batch's convergence time is fully attributed.
+        """
         paths = tuple(dict.fromkeys(files))
         if not paths:
             return {}, {}
@@ -366,10 +383,15 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState],
             active_paths = tuple(path for path in paths if path not in blocked_paths)
             if not active_paths:
                 continue
+            if stage.whole_archive and not whole_archive:
+                for path in active_paths:
+                    self._file_states[path].stages[stage_name] = StageState.SKIPPED
+                continue
 
             if stage.check_many is None or stage.execute_many is None:
                 for path in active_paths:
                     state = self._file_states[path]
+                    t_check = time.perf_counter()
                     try:
                         needs_work = stage.check(path)
                     except Exception:
@@ -382,6 +404,8 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState],
                         state.stages[stage_name] = StageState.FAILED
                         state.error_count += 1
                         continue
+                    finally:
+                        _record_stage_times(batch_stage_times, f"{stage_name}.check", time.perf_counter() - t_check, {})
 
                     if not needs_work:
                         state.stages[stage_name] = StageState.DONE
@@ -419,6 +443,7 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState],
                         scope="stage",
                     )
             else:
+                t_check = time.perf_counter()
                 try:
                     batch_needs_work = set(stage.check_many(active_paths)).intersection(active_paths)
                 except Exception:
@@ -428,6 +453,7 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState],
                         state.stages[stage_name] = StageState.FAILED
                         state.error_count += 1
                 else:
+                    _record_stage_times(batch_stage_times, f"{stage_name}.check", time.perf_counter() - t_check, {})
                     for path in active_paths:
                         if path not in batch_needs_work:
                             self._file_states[path].stages[stage_name] = StageState.DONE

polylogue/daemon/convergence_stages.py

@@ -426,6 +426,7 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn:
         execute=execute,
         check_many=check_many,
         execute_many=execute_many,
+        whole_archive=True,
     )
 
 
@@ -480,6 +481,7 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn:
         execute=execute,
         check_many=check_many,
         execute_many=execute_many,
+        whole_archive=True,
     )
 
 
@@ -539,7 +541,6 @@ def execute(path: Path) -> StageExecuteReturn:
                     session_ids=session_ids,
                     page_size=_DAEMON_INSIGHT_REBUILD_PAGE_SIZE,
                 )
-                _record_fts_freshness_after_insights(conn)
                 conn.commit()
                 logger.info(
                     "insights: refreshed sessions=%d profiles=%d work_events=%d phases=%d threads=%d",
@@ -614,7 +615,6 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn:
                     session_ids=session_ids,
                     page_size=_DAEMON_INSIGHT_REBUILD_PAGE_SIZE,
                 )
-                _record_fts_freshness_after_insights(conn)
                 conn.commit()
                 logger.info(
                     "insights: batch refreshed paths=%d sessions=%d profiles=%d work_events=%d phases=%d threads=%d",
@@ -681,7 +681,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn:
                     session_ids=ids,
                     page_size=_DAEMON_INSIGHT_REBUILD_PAGE_SIZE,
                 )
-                _record_fts_freshness_after_insights(conn)
                 conn.commit()
                 remaining = _stale_session_profile_ids(conn, ids)
                 logger.info(
@@ -1067,6 +1066,70 @@ def execute_many(_paths: Sequence[Path]) -> StageExecuteReturn:
         check_many=check_many,
         execute_many=execute_many,
         false_means_pending=True,
+        whole_archive=True,
+    )
+
+
+def make_fts_readiness_stage(db_path: Path) -> ConvergenceStage:
+    """Publish the exact archive-wide FTS readiness audit once per pass.
+
+    The audit aggregates every ``blocks`` and ``messages_fts`` row, so it is
+    whole-archive work: partition-scoped FTS repair and per-session insight
+    rebuilds leave it to this stage, which runs after both, once per
+    whole-archive convergence pass.
+    """
+
+    def archive_db() -> Path:
+        return _active_archive_index_path(db_path) or db_path
+
+    def publish() -> StageExecuteReturn:
+        database = archive_db()
+        if not database.exists():
+            return True
+        try:
+            conn = _open_archive_insight_write_connection(database)
+            try:
+                _record_fts_freshness_after_insights(conn)
+                conn.commit()
+            finally:
+                conn.close()
+        except Exception as exc:
+            if _is_transient_sqlite_lock(exc):
+                logger.info("fts readiness: audit deferred because sqlite is busy: %s", exc)
+                return False
+            logger.warning("fts readiness: audit failed", exc_info=True)
+            raise
+        return True
+
+    def check(path: Path) -> bool:
+        return archive_db().exists()
+
+    def execute(path: Path) -> StageExecuteReturn:
+        return publish()
+
+    def check_many(paths: Sequence[Path]) -> set[Path]:
+        return set(paths) if paths and archive_db().exists() else set()
+
+    def execute_many(paths: Sequence[Path]) -> StageExecuteReturn:
+        return publish() if paths else True
+
+    def check_sessions(session_ids: Sequence[str]) -> set[str]:
+        return set(session_ids) if session_ids and archive_db().exists() else set()
+
+    def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn:
+        return publish() if session_ids else True
+
+    return ConvergenceStage(
+        name="fts_readiness",
+        description="Publish the exact archive-wide FTS readiness audit",
+        check=check,
+        execute=execute,
+        check_many=check_many,
+        execute_many=execute_many,
+        check_sessions=check_sessions,
+        execute_sessions=execute_sessions,
+        false_means_pending=True,
+        whole_archive=True,
     )
 
 
@@ -1107,6 +1170,7 @@ def make_default_convergence_stages(
             make_claude_workflow_stage(db_path),
             make_delegation_work_evidence_stage(db_path),
             make_derived_stage(db_path),
+            make_fts_readiness_stage(db_path),
             make_standing_query_stage(db_path, evaluator=ArchiveCanonicalPlanEvaluator(db_path)),
         )
     )
@@ -2449,9 +2513,8 @@ def _archive_insights_execute_ids(
     finally:
         if marker_conn is not None:
             marker_conn.close()
-    # The rebuild commits its own rows. Publish and commit the final exact FTS
-    # state in the same production stage before reporting success.
-    _record_fts_freshness_after_insights(conn)
+    # The rebuild commits its own rows; the exact archive-wide FTS audit is
+    # published once per whole-archive pass by ``make_fts_readiness_stage``.
     conn.commit()
     remaining = _archive_stale_session_profile_ids(conn, list(session_ids))
     logger.info(
@@ -2474,6 +2537,7 @@ def _archive_insights_execute_ids(
     "make_delegation_work_evidence_stage",
     "make_default_convergence_stages",
     "make_embed_stage",
+    "make_fts_readiness_stage",
     "make_fts_stage",
     "make_derived_stage",
     "make_raw_authority_verdict_cache_stage",

polylogue/daemon/fts_convergence.py

@@ -80,7 +80,11 @@ def run_once_sync(
                 )
             with open_daemon_connection(self._db_path, timeout=30.0) as conn:
                 result = FtsDerivationAdapter().converge(conn, keys=partition_keys)
-                if result.outcome is FtsOutcome.DONE:
+                # The readiness projection is an exact archive-wide audit;
+                # a partition-scoped pass cannot claim it and must not pay
+                # for it (``make_fts_readiness_stage`` publishes it once per
+                # whole-archive convergence pass).
+                if result.outcome is FtsOutcome.DONE and partition_keys is None:
                     self._publish_readiness_projection(conn)
             state = {
                 FtsOutcome.DONE: FtsOwnerState.READY_EXACT,

polylogue/sources/live/batch.py

@@ -759,8 +759,14 @@ async def ingest_files(
         skipped_file_count: int = 0,
         emit_event: bool = True,
         max_pass_seconds: float | None = None,
+        whole_archive_convergence: bool = True,
     ) -> LiveBatchMetrics:
-        """Ingest files in batch, run post-ingest convergence, and return metrics."""
+        """Ingest files in batch, run post-ingest convergence, and return metrics.
+
+        ``whole_archive_convergence=False`` bounds post-ingest convergence to
+        this batch's own subjects (a catch-up chunk); the caller runs one
+        whole-archive pass at the end of its catch-up.
+        """
         authorization = self.require_cursor_authority(paths)
         refused_paths = self._refused_paths
         self._refused_paths = frozenset()
@@ -906,6 +912,8 @@ async def flush_append_plans() -> None:
                 "watcher.live_ingest.append_convergence",
                 self._converge_paths,
                 [plan.path for plan in append_result.succeeded],
+                whole_archive=whole_archive_convergence,
+                session_ids=tuple(append_result.session_ids_by_path.values()),
             )
             convergence_time_s += elapsed
             release_process_memory()
@@ -1156,6 +1164,8 @@ async def flush_append_plans() -> None:
                         "watcher.live_ingest.full_convergence",
                         self._converge_paths,
                         full_result.succeeded,
+                        whole_archive=whole_archive_convergence,
+                        session_ids=full_result.changed_session_ids,
                     )
                     convergence_time_s += elapsed
                     release_process_memory()
@@ -1733,7 +1743,11 @@ def _record_convergence_outcome(self, path: Path, debts: Iterable[ConvergenceDeb
         record_convergence_outcome(self._cursor, path, debts, archive_root=archive_root)
 
     def _converge_paths(
-        self, paths: Iterable[Path]
+        self,
+        paths: Iterable[Path],
+        *,
+        whole_archive: bool = True,
+        session_ids: Iterable[str] = (),
     ) -> tuple[set[Path], float, dict[str, float], list[ConvergenceDebt]]:
         unique_paths = tuple(sorted(dict.fromkeys(paths)))
         if not unique_paths:
@@ -1745,23 +1759,33 @@ def _converge_paths(
         try:
             converge_batch = getattr(self._converger, "converge_batch", None)
             if callable(converge_batch):
-                states, timings = converge_batch(unique_paths)
+                # The keyword is passed only when it narrows the pass, so
+                # convergers without the parameter keep their whole-archive
+                # default.
+                states, timings = (
+                    converge_batch(unique_paths) if whole_archive else converge_batch(unique_paths, whole_archive=False)
+                )
                 batch_completed = {
                     path for path in unique_paths if path in states and bool(getattr(states[path], "converged", False))
                 }
                 debt_items = convergence_debt_from_states(unique_paths, states)
-                # #1654: after convergence, check for new hook events
-                # that carry paste evidence and update matching messages.
+                batch_stage_timings = {stage_name: float(elapsed) for stage_name, elapsed in timings.items()}
+                # #1654: after convergence, check for new hook events that
+                # carry paste evidence and update matching messages. Scoped to
+                # this batch's sessions so the scan is bounded by the batch,
+                # not by the archive's whole hook history.
+                t_paste = time.perf_counter()
                 try:
                     from polylogue.sources.live.hook_paste_enrichment import enrich_paste_from_hooks
 
-                    enrich_paste_from_hooks(self._cursor._db_path)
+                    enrich_paste_from_hooks(self._cursor._db_path, session_ids=tuple(dict.fromkeys(session_ids)))
                 except Exception:
                     logger.debug("hook_paste: enrichment failed (non-fatal)", exc_info=True)
+                batch_stage_timings["hook_paste_enrichment"] = time.perf_counter() - t_paste
                 return (
                     batch_completed,
                     time.perf_counter() - started,
-                    {stage_name: float(elapsed) for stage_name, elapsed in timings.items()},
+                    batch_stage_timings,
                     debt_items,
                 )