refactor(daemon): delete dead converger process-pool wiring

CLAUDE.md

@@ -155,8 +155,8 @@ deliberate tricks:
 - Hot-file quiet deferral (`convergence_stages.py`) batches still-appending
   Codex/Claude sessions until a quiet window; embed runs in bounded windows.
 
-CPU-bound stages go to a `ProcessPoolExecutor`; the main process stays the
-**sole SQLite writer**. Blob GC uses two independent safety invariants (leases +
+The main process is the **sole SQLite writer** — no convergence stage runs in
+a worker process. Blob GC uses two independent safety invariants (leases +
 snapshot reference check) to bridge the acquire-blob → commit-row window.
 
 ### Schema regimes (durability-keyed)

devtools/daemon_live_benchmark.py

@@ -212,25 +212,21 @@ async def run_daemon_live_convergence_workload(db_path: Path) -> tuple[dict[str,
 
     scale = scale_from_db_path(db_path)
     workload = generate_daemon_live_workload(db_path.parent, scale=scale)
-    converger = DaemonConverger(stages=make_default_convergence_stages(db_path), max_workers=2)
-    await converger.start()
-    try:
-        async with Polylogue(archive_root=db_path.parent, db_path=db_path) as polylogue:
-            processor = LiveBatchProcessor(
-                polylogue,
-                tuple(WatchSource(name=source.name, root=source.root) for source in workload.sources),
-                cursor=CursorStore(db_path),
-                parser_fingerprint="daemon-live-benchmark-v1",
-                converger=converger,
-            )
-            initial_metrics = await processor.ingest_files(workload.files, emit_event=False)
-            workload = append_daemon_live_workload(
-                workload,
-                message_index=DAEMON_LIVE_SCALE_SPECS.get(scale, DAEMON_LIVE_SCALE_SPECS["small"]).messages_per_file,
-            )
-            append_metrics = await processor.ingest_files(workload.files, emit_event=False)
-    finally:
-        await converger.stop()
+    converger = DaemonConverger(stages=make_default_convergence_stages(db_path))
+    async with Polylogue(archive_root=db_path.parent, db_path=db_path) as polylogue:
+        processor = LiveBatchProcessor(
+            polylogue,
+            tuple(WatchSource(name=source.name, root=source.root) for source in workload.sources),
+            cursor=CursorStore(db_path),
+            parser_fingerprint="daemon-live-benchmark-v1",
+            converger=converger,
+        )
+        initial_metrics = await processor.ingest_files(workload.files, emit_event=False)
+        workload = append_daemon_live_workload(
+            workload,
+            message_index=DAEMON_LIVE_SCALE_SPECS.get(scale, DAEMON_LIVE_SCALE_SPECS["small"]).messages_per_file,
+        )
+        append_metrics = await processor.ingest_files(workload.files, emit_event=False)
 
     total_wall_s = initial_metrics.total_time_s + append_metrics.total_time_s
     files_per_s = (

polylogue/daemon/cli.py

@@ -1147,7 +1147,7 @@ def _drain_convergence_debt_once(db: Path, *, limit: int = 100) -> int:
             session_ids = tuple(
                 dict.fromkeys(debt.subject_id for debt in stage_debt if debt.subject_type == "session_id")
             )
-            converger = DaemonConverger(stages=selected_stages, max_workers=2)
+            converger = DaemonConverger(stages=selected_stages)
             path_states, _path_timings = converger.converge_batch(paths)
             session_states, _session_timings = converger.converge_sessions(session_ids)
             subject_states.update(
@@ -1728,9 +1728,7 @@ async def run_daemon_services(
                     _db,
                     embed_defer=(lambda: embed_gate is not None and not embed_gate.is_set()),
                 ),
-                max_workers=2,
             )
-            await converger.start()
             if lifecycle_events_enabled:
                 await _emit_daemon_lifecycle_event(
                     "component_started",
@@ -1821,21 +1819,14 @@ async def run_daemon_services(
                 )
             if watcher is not None:
                 watcher.stop()
-            if converger is not None:
-                try:
-                    async with asyncio.timeout(5.0):
-                        await converger.stop()
-                except TimeoutError:
-                    logger.warning("daemon: timed out stopping convergence executor")
             if _daemon_parse_stage_singleton is not None:
                 # polylogue-m6tp phase (a), CodeRabbit PR #3168: the parse-stage
                 # warmer's ThreadPoolExecutor is created lazily only when
                 # daemon_parse_stage_split is enabled, and otherwise never
                 # touched here. shutdown() is non-blocking (wait=False,
-                # cancel_futures=True) so no timeout wrapper is needed -- unlike
-                # converger.stop() above, it cannot itself hang the shutdown
-                # sequence; it just stops the pool from keeping the process
-                # alive at exit.
+                # cancel_futures=True) so no timeout wrapper is needed: it
+                # cannot itself hang the shutdown sequence; it just stops the
+                # pool from keeping the process alive at exit.
                 _daemon_parse_stage_singleton.shutdown()
             if server is not None:
                 await _shutdown_server_if_serving(server, server_task, label="browser-capture")

polylogue/daemon/convergence.py

@@ -16,13 +16,11 @@
 
 import time
 from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
-from concurrent.futures import ProcessPoolExecutor
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
 
 from polylogue.logging import get_logger
-from polylogue.pipeline.services.process_pool import process_pool_executor, terminate_process_pool
 
 logger = get_logger(__name__)
 _INSIGHT_DEFERRED_UNTIL_QUIET = "insights deferred until source quiet"
@@ -76,8 +74,6 @@ class ConvergenceStage:
     # These avoid resolving a failed derived subject back to source files.
     check_sessions: Callable[[Sequence[str]], set[str]] | None = None
     execute_sessions: Callable[[Sequence[str]], StageExecuteReturn] | None = None
-    # Can run in a worker process (CPU-bound, no SQLite write).
-    cpu_bound: bool = False
     # Some stages intentionally return False after doing bounded successful
     # work so the remaining backlog is retried as convergence debt.
     false_means_pending: bool = False
@@ -167,29 +163,18 @@ class DaemonConverger:
     """Drives archive state toward desired state for all source files.
 
     Runs a set of :class:`ConvergenceStage` checks against each file.
-    CPU-bound stages are dispatched to a :class:`~concurrent.futures.ProcessPoolExecutor`.
     The main process is the only SQLite writer.
     """
 
-    def __init__(
-        self,
-        stages: Iterable[ConvergenceStage],
-        *,
-        max_workers: int | None = None,
-    ) -> None:
+    def __init__(self, stages: Iterable[ConvergenceStage]) -> None:
         self._stages: dict[str, ConvergenceStage] = {s.name: s for s in stages}
-        self._max_workers = max_workers or 2
         self._file_states: dict[Path, FileState] = {}
         self._session_states: dict[str, SessionState] = {}
-        self._executor: ProcessPoolExecutor | None = None
 
     @property
     def stage_names(self) -> list[str]:
         return list(self._stages)
 
-    def _has_cpu_bound_stage(self) -> bool:
-        return any(stage.cpu_bound for stage in self._stages.values())
-
     def stage_status(self) -> dict[str, dict[str, object]]:
         """Return bounded stage-owned status without propagating secret detail."""
         result: dict[str, dict[str, object]] = {}
@@ -278,29 +263,6 @@ def _session_barriers_blocked(
             return set(session_ids)
         return blocked.intersection(session_ids)
 
-    async def start(self) -> None:
-        if self._executor is not None:
-            return
-        if not self._has_cpu_bound_stage():
-            logger.info(
-                "converger: started without worker pool, stages=%s",
-                list(self._stages),
-            )
-            return
-        self._executor = process_pool_executor(max_workers=self._max_workers)
-        logger.info(
-            "converger: started with %d worker(s), stages=%s",
-            self._max_workers,
-            list(self._stages),
-        )
-
-    async def stop(self) -> None:
-        executor = self._executor
-        self._executor = None
-        if executor is not None:
-            terminate_process_pool(executor)
-        logger.info("converger: stopped")
-
     def converge_file(self, path: Path) -> FileState:
         """Converge one file while honoring durable stage barriers."""
         if path not in self._file_states:
@@ -334,11 +296,7 @@ def converge_file(self, path: Path) -> FileState:
                         state.stages[stage_name] = StageState.IN_PROGRESS
                         t_stage = time.perf_counter()
                         try:
-                            if stage.cpu_bound and self._executor is not None:
-                                future = self._executor.submit(stage.execute, path)
-                                execute_result = future.result()
-                            else:
-                                execute_result = stage.execute(path)
+                            execute_result = stage.execute(path)
                         except Exception as exc:
                             logger.warning(
                                 "converger: execute failed for %s stage=%s: %s",
@@ -412,7 +370,7 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState],
             if not active_paths:
                 continue
 
-            if stage.check_many is None or stage.execute_many is None or stage.cpu_bound:
+            if stage.check_many is None or stage.execute_many is None:
                 for path in active_paths:
                     state = self._file_states[path]
                     try:
@@ -561,7 +519,7 @@ def converge_sessions(
             if not active_ids:
                 continue
 
-            if stage.check_sessions is None or stage.execute_sessions is None or stage.cpu_bound:
+            if stage.check_sessions is None or stage.execute_sessions is None:
                 for session_id in active_ids:
                     self._session_states[session_id].stages[stage_name] = StageState.SKIPPED
             else:

polylogue/daemon/convergence_stages.py

@@ -257,7 +257,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn:
         execute_many=execute_many,
         check_sessions=check_sessions,
         execute_sessions=execute_sessions,
-        cpu_bound=False,
         false_means_pending=True,
     )
 
@@ -337,7 +336,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn:
         execute_many=execute_many,
         check_sessions=check_sessions,
         execute_sessions=execute_sessions,
-        cpu_bound=False,
         false_means_pending=True,
     )
 
@@ -406,7 +404,6 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn:
         execute=execute,
         check_many=check_many,
         execute_many=execute_many,
-        cpu_bound=False,
     )
 
 
@@ -641,7 +638,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn:
         execute_many=execute_many,
         check_sessions=check_sessions,
         execute_sessions=execute_sessions,
-        cpu_bound=False,
         false_means_pending=True,
     )
 
@@ -759,7 +755,6 @@ def barrier_many(paths: Sequence[Path]) -> set[Path]:
         execute_many=execute_many,
         check_sessions=check_sessions,
         execute_sessions=execute_sessions,
-        cpu_bound=False,
         false_means_pending=True,
         blocks_following_stages=service.mode is PublicationMode.PRIMARY,
         barrier_check=barrier,

polylogue/daemon/convergence_standing_queries.py

@@ -117,7 +117,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn:
         execute=execute,
         check_sessions=check_sessions,
         execute_sessions=execute_sessions,
-        cpu_bound=False,
         false_means_pending=True,
     )
 

tests/benchmarks/test_daemon_convergence.py

@@ -128,7 +128,7 @@ def _run_convergence_probe(
     # Collect all JSONL files.
     files = list(corpus_root.rglob("*.jsonl"))
 
-    converger = DaemonConverger(stages=make_default_convergence_stages(db_path), max_workers=4)
+    converger = DaemonConverger(stages=make_default_convergence_stages(db_path))
     polylogue = _BenchmarkPolylogue(tmp_path, db_path)
     processor = LiveBatchProcessor(
         cast(Any, polylogue),
@@ -264,7 +264,7 @@ def _run_convergence_memory_probe(
 
     files = list(corpus_root.rglob("*.jsonl"))
 
-    converger = DaemonConverger(stages=make_default_convergence_stages(db_path), max_workers=4)
+    converger = DaemonConverger(stages=make_default_convergence_stages(db_path))
     polylogue = _BenchmarkPolylogue(tmp_path, db_path)
     processor = LiveBatchProcessor(
         cast(Any, polylogue),

tests/benchmarks/test_daemon_convergence_multi_provider.py

@@ -71,7 +71,7 @@ def _run_convergence_probe(
     # Filter only session files (skip metadata)
     files = [f for f in files if not f.name.startswith(".")]
 
-    converger = DaemonConverger(stages=make_default_convergence_stages(db_path), max_workers=4)
+    converger = DaemonConverger(stages=make_default_convergence_stages(db_path))
     polylogue = _BenchmarkPolylogue(tmp_path, db_path)
     processor = LiveBatchProcessor(
         cast(Any, polylogue),

tests/integration/test_daemon_convergence_evidence.py

@@ -181,10 +181,7 @@ def test_daemon_convergence_evidence_full_archive_state(
     assert before["fts_trigger_state"]["all_present"] is True, before["fts_trigger_state"]
 
     # ── Drive convergence: same primitives as polylogued run ─────────
-    converger = DaemonConverger(
-        stages=make_default_convergence_stages(db_path),
-        max_workers=2,
-    )
+    converger = DaemonConverger(stages=make_default_convergence_stages(db_path))
     polylogue = _MinimalPolylogue(tmp_path, db_path)
     processor = LiveBatchProcessor(
         cast(Any, polylogue),
@@ -194,15 +191,7 @@ def test_daemon_convergence_evidence_full_archive_state(
         converger=converger,
     )
 
-    async def _run_convergence() -> Any:
-        await converger.start()
-        try:
-            metrics = await processor.ingest_files(files, emit_event=False)
-        finally:
-            await converger.stop()
-        return metrics
-
-    metrics = asyncio.run(_run_convergence())
+    metrics = asyncio.run(processor.ingest_files(files, emit_event=False))
 
     # ── Ingest completeness ─────────────────────────────────────────
     assert metrics.failed_file_count == 0, (

tests/unit/daemon/test_convergence_final_state.py

@@ -93,10 +93,7 @@ def test_convergence_produces_consistent_final_archive_state(
         _write_claude_code_session(p, session_id, n_msgs)
         files.append(p)
 
-    converger = DaemonConverger(
-        stages=make_default_convergence_stages(db_path),
-        max_workers=2,
-    )
+    converger = DaemonConverger(stages=make_default_convergence_stages(db_path))
     polylogue = _MinimalPolylogue(tmp_path, db_path)
     processor = LiveBatchProcessor(
         cast(Any, polylogue),

tests/unit/daemon/test_daemon_cli.py

@@ -3454,14 +3454,8 @@ async def fake_loop(name: str) -> None:
 
     class FakeConverger:
         def __init__(self, *_args: object, **_kwargs: object) -> None:
-            pass
-
-        async def start(self) -> None:
             events.append("converger")
 
-        async def stop(self) -> None:
-            events.append("converger-stop")
-
     class FakeAPIServer:
         def __init__(self) -> None:
             self.stopped = threading.Event()
@@ -3742,11 +3736,7 @@ def server_close(self) -> None:
             self.close_called = True
 
     class FakeConverger:
-        async def start(self) -> None:
-            return None
-
-        async def stop(self) -> None:
-            return None
+        pass
 
     async def noop() -> None:
         return None