fix(daemon): a fresh daemon converges its backlog on a live host

polylogue/daemon/cli.py

@@ -2016,7 +2016,9 @@ def _browser_capture_spool_has_pending_files() -> bool:
     if not spool.exists():
         return False
     now = time.time()
-    cursor_store = CursorStore(_active_index_db_path())
+    # Read-only probe: the writer initialized the ops tier at startup, and
+    # re-initializing here would write against the writer lease.
+    cursor_store = CursorStore(_active_index_db_path(), initialize=False)
     for path in spool.rglob("*.json"):
         try:
             stat = path.stat()

polylogue/daemon/status.py

@@ -310,6 +310,7 @@ class RawFrontierIntegrity(BaseModel):
     cursor_ahead_samples: list[dict[str, object]] = Field(default_factory=list)
     cursor_authority_gap_count: int = 0
     cursor_authority_gap_samples: list[dict[str, object]] = Field(default_factory=list)
+    cursor_authority_deferred_count: int = 0
     cursor_ahead_reason: str = ""
 
 

polylogue/readiness/__init__.py

@@ -644,6 +644,7 @@ def _raw_frontier_integrity_check(projection: RawFrontierIntegrityProjection) ->
         "cursor_head_comparison_count": projection.cursor_head_comparison_count,
         "cursor_ahead_comparison_count": projection.cursor_ahead_comparison_count,
         "cursor_authority_gap_count": projection.cursor_authority_gap_count,
+        "cursor_authority_deferred_count": projection.cursor_authority_deferred_count,
     }
     issue_count = (
         projection.broken_head_count

polylogue/readiness/capability.py

@@ -318,6 +318,7 @@ def component_from_raw_frontier_integrity(payload: Mapping[str, Any] | None) ->
             "cursor_head_comparison_count": _raw_frontier_count(data.get("cursor_head_comparison_count")),
             "cursor_ahead_comparison_count": _raw_frontier_count(data.get("cursor_ahead_comparison_count")),
             "cursor_authority_gap_count": _raw_frontier_count(data.get("cursor_authority_gap_count")),
+            "cursor_authority_deferred_count": _raw_frontier_count(data.get("cursor_authority_deferred_count")),
         },
         caveats=tuple(caveats),
         repair_hint=None if state == CapabilityReadinessState.READY else "polylogue ops status --full",
@@ -377,6 +378,7 @@ def unknown_raw_frontier_integrity_projection(reason: str) -> RawFrontierIntegri
     "cursor_head_comparison_count",
     "cursor_ahead_comparison_count",
     "cursor_authority_gap_count",
+    "cursor_authority_deferred_count",
 )
 _RAW_FRONTIER_SAMPLE_COUNT_KEYS = (
     ("broken_head_samples", "broken_head_count"),
@@ -401,6 +403,9 @@ def _validate_raw_frontier_projection(
 ) -> tuple[dict[str, Any] | None, str | None]:
     """Validate and aggregate one complete canonical frontier projection."""
 
+    # The deferred count is informational: an older producer that omits it
+    # still describes a complete projection.
+    payload = {"cursor_authority_deferred_count": 0, **payload}
     missing = required_keys.difference(payload)
     if missing:
         return None, f"missing field(s): {', '.join(sorted(missing))}"
@@ -498,9 +503,10 @@ def raw_frontier_source_selection_block_reason(
     Raw convergence and reindex both select source rows before they write an
     index. They therefore require the same complete, healthy frontier proof
     exposed by the diagnostic/readiness route. ``None`` means every authority
-    comparison was proven healthy. A deferred or incomparable cursor is a
-    deliberate terminal exception to the byte comparison, but it remains a
-    blocking unknown until its authority can be resolved.
+    comparison was proven healthy. An incomparable cursor is a blocking
+    unknown until its authority can be resolved; a deferred cursor (a durably
+    captured tail awaiting the quiet window) is a typed safe state and never
+    blocks selection of other paths.
     """
 
     projection = raw_frontier_integrity_projection(

polylogue/sources/live/batch.py

@@ -198,6 +198,7 @@
 
 if TYPE_CHECKING:
     from polylogue.api import Polylogue
+    from polylogue.storage.raw_retention import RawFrontierBlockedPaths
 
 logger = get_logger(__name__)
 
@@ -585,6 +586,7 @@ def __init__(
         sync_runner: LiveBatchSyncRunner | None = None,
         parse_stage: LiveParseStage | None = None,
     ) -> None:
+        self._refused_paths: frozenset[Path] = frozenset()
         self._polylogue = polylogue
         self._sources = tuple(sources)
         self._cursor = cursor
@@ -700,8 +702,55 @@ def require_cursor_authority(self, paths: Iterable[Path] | None = None) -> Curso
             for path in paths
         ):
             return None
+        # The proof names the paths it refuses. Refusing only those keeps one
+        # anomalous file from stalling the other 45,000; a refusal that no
+        # path explains still blocks everything.
+        blocked = self._blocked_source_paths()
+        if blocked.unattributed_reason is None:
+            if paths is None:
+                logger.warning(
+                    "live.watcher: cursor authority refuses %d source path(s); path-less route proceeds: %s",
+                    len(blocked.source_paths),
+                    reason,
+                )
+                return None
+            selected = [Path(path) for path in paths]
+            refused = frozenset(
+                path
+                for path in selected
+                if str(path) in blocked.source_paths or str(path.resolve()) in blocked.source_paths
+            )
+            if len(refused) < len(selected):
+                self._refused_paths = refused
+                if refused:
+                    logger.warning(
+                        "live.watcher: cursor authority refused %d of %d path(s) in this batch: %s",
+                        len(refused),
+                        len(selected),
+                        reason,
+                    )
+                return None
         raise CursorAuthorityBlockedError(f"live watcher source-selection gate blocked: {reason}")
 
+    def admit_paths(self, paths: Iterable[Path]) -> list[Path]:
+        """The subset of ``paths`` the frontier proof admits, in order.
+
+        Raises when nothing may proceed: every path is refused, or the refusal
+        is one no path explains.
+        """
+        selected = list(paths)
+        self.require_cursor_authority(selected)
+        refused = self._refused_paths
+        self._refused_paths = frozenset()
+        return [path for path in selected if path not in refused]
+
+    def _blocked_source_paths(self) -> RawFrontierBlockedPaths:
+        archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent))
+        from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot
+        from polylogue.storage.raw_retention import raw_frontier_blocked_source_paths
+
+        return raw_frontier_blocked_source_paths(archive_root, raw_materialization_readiness_snapshot(archive_root))
+
     async def ingest_files(
         self,
         paths: list[Path],
@@ -713,6 +762,11 @@ async def ingest_files(
     ) -> LiveBatchMetrics:
         """Ingest files in batch, run post-ingest convergence, and return metrics."""
         authorization = self.require_cursor_authority(paths)
+        refused_paths = self._refused_paths
+        self._refused_paths = frozenset()
+        if refused_paths:
+            paths = [path for path in paths if path not in refused_paths]
+            skipped_file_count += len(refused_paths)
         if is_fully_degraded():
             # The daemon has been marked structurally unable to ingest (e.g.
             # schema mismatch detected at preflight or on the first batch).

polylogue/sources/live/watcher.py

@@ -266,6 +266,12 @@ class CatchUpPlan:
     needed_bytes: int
 
 
+def _is_retryable_lock_error(exc: sqlite3.OperationalError) -> bool:
+    """SQLite lock contention, as opposed to a broken database."""
+    message = str(exc).lower()
+    return "database is locked" in message or "database table is locked" in message or "busy" in message
+
+
 class LiveWatcher:
     """Async watcher that ingests grown JSONL files in batches.
 
@@ -598,11 +604,13 @@ async def prepare_catch_up() -> None:
             # planning pass. ``_plan_catch_up`` can reconcile missing cursors
             # and rebase matching filesystem observations before ingestion has
             # a chance to apply its own gate.
-            self._batch_processor.require_cursor_authority()
+            admitted = self._batch_processor.admit_paths([candidate.path for candidate in candidates])
             await self._run_writer_sync("watcher.catch_up.cursor_initialize", self._cursor.initialize)
-            self._batch_processor.require_cursor_authority()
-            logger.info("live.watcher: catch-up scan over %d file(s)", len(candidates))
-            plan_holder.append(self._plan_catch_up(candidates))
+            admitted = self._batch_processor.admit_paths(admitted)
+            admitted_set = set(admitted)
+            planned = tuple(candidate for candidate in candidates if candidate.path in admitted_set)
+            logger.info("live.watcher: catch-up scan over %d file(s)", len(planned))
+            plan_holder.append(self._plan_catch_up(planned))
 
         try:
             await self._run_coordinated("watcher.catch_up.prefilter", prepare_catch_up)
@@ -688,7 +696,23 @@ async def ingest_chunk(
                         ):
                             self._defer_unaccounted_failed_retries(chunk_paths)
 
-                await self._run_coordinated("watcher.catch_up.chunk", ingest_chunk)
+                try:
+                    await self._run_coordinated("watcher.catch_up.chunk", ingest_chunk)
+                except sqlite3.OperationalError as exc:
+                    if not _is_retryable_lock_error(exc):
+                        raise
+                    # A write that lost a lock race is this chunk's failure,
+                    # never the daemon's death: the cursor retry policy brings
+                    # the chunk back. Rehearsal 2026-09-05 died here while the
+                    # Drive catch-up held source.db for 112 s.
+                    failed += len(chunk_paths)
+                    logger.warning(
+                        "live.watcher: catch-up chunk %d/%d deferred, archive write lost a lock race: %s",
+                        chunk_index,
+                        len(chunks),
+                        exc,
+                    )
+                    self._defer_unaccounted_failed_retries(chunk_paths)
             if self._stop.is_set():
                 await self._emit_catch_up_terminal(
                     operation_id, "stopped", plan, attempted, ingested, failed, stage_timings_s, cycle_started
@@ -1008,12 +1032,13 @@ async def _flush_pending(self) -> bool:
             self._forced_reparse_paths.difference_update(paths)
 
         async def flush_batch() -> None:
+            nonlocal paths
             # Filtering a changed-file batch invokes cursor reconciliation and
             # lifecycle actuators, so the source-selection proof must be
             # consumed before initialization or any stateful decision.
-            self._batch_processor.require_cursor_authority()
+            paths = self._batch_processor.admit_paths(paths)
             await self._run_writer_sync("watcher.live_batch.cursor_initialize", self._cursor.initialize)
-            self._batch_processor.require_cursor_authority()
+            paths = self._batch_processor.admit_paths(paths)
             # Filter to files that actually need work.
             cursor_records = self._cursor.get_records(paths)
             needed = []
@@ -1032,11 +1057,18 @@ async def flush_batch() -> None:
                 return
 
             logger.info("live.watcher: batching %d changed file(s)", len(needed))
-            metrics = await self._ingest_files(
-                needed,
-                queued_file_count=len(paths),
-                skipped_file_count=len(paths) - len(needed),
-            )
+            try:
+                metrics = await self._ingest_files(
+                    needed,
+                    queued_file_count=len(paths),
+                    skipped_file_count=len(paths) - len(needed),
+                )
+            except sqlite3.OperationalError as exc:
+                if not _is_retryable_lock_error(exc):
+                    raise
+                logger.warning("live.watcher: changed-file batch deferred, archive write lost a lock race: %s", exc)
+                self._defer_unaccounted_failed_retries(needed)
+                return
             if metrics is not None:
                 _log_ingest_metrics("live.watcher: changed-file batch", metrics)
                 if (
@@ -1671,7 +1703,7 @@ async def _ingest_files(
         skipped_file_count: int = 0,
     ) -> LiveBatchMetrics:
         """Ingest files through the reusable daemon live batch processor."""
-        self._batch_processor.require_cursor_authority()
+        self._batch_processor.require_cursor_authority(paths)
         async with self._ingest_lock:
 
             async def ingest() -> LiveBatchMetrics: