From 856a61d5cedaed676017e07a3bb603e582b3d62c Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 19 Jul 2026 22:15:27 +0200 Subject: [PATCH 1/2] refactor(daemon): delete dead converger process-pool wiring Problem: no production ConvergenceStage ever sets cpu_bound=True (the only True was a synthetic test fixture), so DaemonConverger's start()/stop() pool lifecycle, _has_cpu_bound_stage(), and the converge_file cpu_bound submit branch were dead machinery. Worse, converge_batch (the primary production path) never honored cpu_bound at all - it always ran stage.execute inline in the daemon's main process regardless of the flag, diverging from converge_file's (unreachable) pool-dispatch branch and from the module/CLAUDE.md docstrings claiming "CPU-bound stages are dispatched to a ProcessPoolExecutor". Solution: delete the pool machinery rather than reconcile the two paths, per the bead's own recommendation (nothing uses it). Removed from polylogue/daemon/convergence.py: the ProcessPoolExecutor/ process_pool_executor/terminate_process_pool imports, the ConvergenceStage.cpu_bound field, DaemonConverger._executor/ _has_cpu_bound_stage, and every stage.cpu_bound branch in converge_file/converge_batch/converge_sessions. start()/stop() remain as trivial async lifecycle hooks (daemon/cli.py awaits them unconditionally and is out of scope for this lane) that now just log. max_workers stays an accepted-but-unused constructor kwarg since daemon/cli.py and several other call sites still pass it positionally. Removed the now-dead cpu_bound=False kwarg from all 6 stage registrations (convergence_stages.py x5, convergence_standing_queries.py x1) and updated CLAUDE.md's daemon section to drop the pool-dispatch claim. Deleted 3 tests in tests/unit/daemon/test_daemon_convergence.py that exercised only the removed pool wiring (start creates/skips a pool, stop cancels pool work) plus their now-unused imports (Future, Mock, pytest). The other 8 tests in that file, which exercise surviving check/execute/batch/session/barrier behavior, are untouched and still pass - that is the proof this is behavior-preserving, per repo test doctrine (no new "assert pool is gone" test was added). Kept out of scope, per the bead and coordinator: pipeline/services/ process_pool.py and its other live users (archive_ingest, validation_flow, ingest_batch, revision_backfill) - owned by the m6tp deletion program; daemon/convergence_stages.py's broader hot-file shape (docs/retro/2026-05-24-1498-cascade.md reviewed first, no hazard for this mechanical edit); daemon/cli.py and daemon/parse_prefetch.py - other lanes are active there. Verification: - devtools test tests/unit/daemon/test_daemon_convergence.py tests/unit/daemon/test_standing_queries.py tests/unit/daemon/test_standing_queries_default_evaluator.py tests/unit/daemon/test_convergence_final_state.py tests/unit/daemon/test_convergence_restart_law.py tests/unit/sinex/test_convergence.py tests/unit/daemon/test_catch_up_observability.py -> "23 passed in 12.61s" - devtools verify --quick -> all 16 steps ok, exit_code 0 (ruff format/check, mypy --strict, render all --check, layering, topology, closure-matrix, schema roundtrip, manifests, ci-workflows, doc-commands, docs-coverage, test-infra-currency, clock-hygiene, pytest-timeout-overrides, degrade-loudly) Ref polylogue-7uqr Co-Authored-By: Claude --- CLAUDE.md | 4 +- polylogue/daemon/convergence.py | 42 ++----- polylogue/daemon/convergence_stages.py | 5 - .../daemon/convergence_standing_queries.py | 1 - tests/unit/daemon/test_daemon_convergence.py | 105 ------------------ 5 files changed, 9 insertions(+), 148 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 971b24a0d5..eee6fd9c5b 100644 --- a/CLAUDE.md +++ b/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) diff --git a/polylogue/daemon/convergence.py b/polylogue/daemon/convergence.py index 235fe5fee9..64737effc5 100644 --- a/polylogue/daemon/convergence.py +++ b/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,7 +163,6 @@ 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. """ @@ -177,19 +172,17 @@ def __init__( *, max_workers: int | None = None, ) -> None: + # max_workers is accepted for caller compatibility but unused: no + # convergence stage runs in a worker process (polylogue-7uqr). + _ = max_workers 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]] = {} @@ -279,26 +272,9 @@ def _session_barriers_blocked( 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), - ) + logger.info("converger: started, stages=%s", 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: @@ -334,11 +310,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 +384,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 +533,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: diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index 56e64e02ba..7068800288 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/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, diff --git a/polylogue/daemon/convergence_standing_queries.py b/polylogue/daemon/convergence_standing_queries.py index d6626b88dd..cb1b6028a3 100644 --- a/polylogue/daemon/convergence_standing_queries.py +++ b/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, ) diff --git a/tests/unit/daemon/test_daemon_convergence.py b/tests/unit/daemon/test_daemon_convergence.py index 481f1ad5de..39f61d0f96 100644 --- a/tests/unit/daemon/test_daemon_convergence.py +++ b/tests/unit/daemon/test_daemon_convergence.py @@ -1,11 +1,7 @@ from __future__ import annotations from collections.abc import Sequence -from concurrent.futures import Future from pathlib import Path -from unittest.mock import Mock - -import pytest from polylogue.daemon.convergence import ConvergenceStage, DaemonConverger, StageExecutionResult, StageState @@ -251,104 +247,3 @@ def check_sessions(session_ids: Sequence[str]) -> set[str]: assert states["conv-b"].stages["embed"] is StageState.PENDING assert states["conv-b"].last_error == "session stage embed returned False" assert set(converger._session_states) == {"conv-b"} - - -@pytest.mark.asyncio -async def test_converger_start_skips_process_pool_for_io_only_stages(monkeypatch: pytest.MonkeyPatch) -> None: - factory = Mock() - monkeypatch.setattr("polylogue.daemon.convergence.process_pool_executor", factory) - - converger = DaemonConverger( - [ - ConvergenceStage( - name="insights", - description="io stage", - check=lambda _candidate: False, - execute=lambda _candidate: True, - ) - ] - ) - - await converger.start() - - factory.assert_not_called() - assert converger._executor is None - - -@pytest.mark.asyncio -async def test_converger_start_creates_process_pool_for_cpu_bound_stages(monkeypatch: pytest.MonkeyPatch) -> None: - class FakeExecutor: - def __init__(self) -> None: - self.shutdown_called: tuple[bool, bool] | None = None - self._processes: dict[int, object] = {} - - def submit(self, func, *args): # type: ignore[no-untyped-def] - future: Future[bool] = Future() - future.set_result(func(*args)) - return future - - def shutdown(self, *, wait: bool = True, cancel_futures: bool = False) -> None: - self.shutdown_called = (wait, cancel_futures) - - fake_executor = FakeExecutor() - factory = Mock(return_value=fake_executor) - monkeypatch.setattr("polylogue.daemon.convergence.process_pool_executor", factory) - - converger = DaemonConverger( - [ - ConvergenceStage( - name="parse", - description="cpu stage", - check=lambda _candidate: False, - execute=lambda _candidate: True, - cpu_bound=True, - ) - ], - max_workers=3, - ) - - await converger.start() - await converger.stop() - - factory.assert_called_once_with(max_workers=3) - assert fake_executor.shutdown_called == (False, True) - - -@pytest.mark.asyncio -async def test_converger_stop_cancels_pending_work_without_waiting() -> None: - class FakeProcess: - def __init__(self) -> None: - self.alive = True - self.terminated = False - - def is_alive(self) -> bool: - return self.alive - - def terminate(self) -> None: - self.terminated = True - self.alive = False - - def join(self, timeout: float | None = None) -> None: - assert timeout is not None - - def kill(self) -> None: - raise AssertionError("cooperative worker should not require SIGKILL") - - class FakeExecutor: - def __init__(self, process: FakeProcess) -> None: - self._processes = {1: process} - self.shutdown_calls: list[tuple[bool, bool]] = [] - - def shutdown(self, *, wait: bool, cancel_futures: bool) -> None: - self.shutdown_calls.append((wait, cancel_futures)) - - process = FakeProcess() - executor = FakeExecutor(process) - converger = DaemonConverger([]) - converger._executor = executor # type: ignore[assignment] - - await converger.stop() - - assert executor.shutdown_calls == [(False, True)] - assert process.terminated is True - assert converger._executor is None From 63f60e176e994de54767c17908f9d6f4a4eb145b Mon Sep 17 00:00:00 2001 From: Sinity Date: Sun, 19 Jul 2026 22:30:21 +0200 Subject: [PATCH 2/2] refactor(daemon): finish converger purge - drop start/stop + max_workers Problem: the first pass of this deletion left start()/stop() as log-only async no-ops (per-caller compatibility with daemon/cli.py, which was out of scope for that lane) and max_workers as an accepted-but-unused constructor kwarg for the same reason. #3168 merged and freed daemon/cli.py, so the vestiges can now go too: just-in-case residue that does no real work should not survive. Solution: - polylogue/daemon/convergence.py: deleted DaemonConverger.start()/ stop() entirely (they only logged; no real lifecycle work remained) and the max_workers constructor parameter. - polylogue/daemon/cli.py: removed the `await converger.start()` call site and the `max_workers=2` kwargs at both DaemonConverger construction sites. Removed the `if converger is not None: try: async with asyncio.timeout(5.0): await converger.stop() except TimeoutError: ...` teardown block, now dead. Left the parse-stage singleton shutdown (#3168, `_daemon_parse_stage_singleton.shutdown()`) and the rest of the teardown sequence (watcher.stop(), server shutdowns) untouched, only rewording its comment's now-stale "unlike converger.stop() above" reference. - devtools/daemon_live_benchmark.py: dropped the `max_workers=2` kwarg and the `await converger.start()`/`try: ... finally: await converger.stop()` wrapper around the benchmark workload. - Updated every other call site that passed `max_workers` or called `.start()`/`.stop()`: tests/benchmarks/test_daemon_convergence.py (x2), tests/benchmarks/test_daemon_convergence_multi_provider.py, tests/integration/test_daemon_convergence_evidence.py (dropped the start/stop wrapper coroutine, calling `processor.ingest_files` directly), tests/unit/daemon/test_convergence_final_state.py, tests/unit/daemon/test_standing_queries.py, tests/unit/daemon/test_standing_queries_default_evaluator.py. - tests/unit/daemon/test_daemon_cli.py: the two `FakeConverger` test doubles patched over `DaemonConverger` no longer need `start`/`stop` methods since cli.py never calls them. One test asserted ordering via `events.index("lineage") < events.index("converger")`, keyed off `FakeConverger.start()` appending "converger" - moved that append into `FakeConverger.__init__` (construction happens at the exact point start() used to fire immediately afterward, so the ordering invariant this protects - converger wired up after lineage readiness, before the watcher - is preserved unchanged). The other FakeConverger's start/stop were unasserted no-ops; reduced to `pass`. Verification: - devtools test tests/unit/daemon/test_daemon_cli.py -> 100 passed - devtools test tests/unit/daemon/test_daemon_convergence.py tests/unit/daemon/test_standing_queries.py tests/unit/daemon/test_standing_queries_default_evaluator.py tests/unit/daemon/test_convergence_final_state.py tests/unit/daemon/test_convergence_restart_law.py tests/unit/sinex/test_convergence.py tests/unit/daemon/test_catch_up_observability.py tests/unit/daemon/test_daemon_cli.py -> 123 passed - devtools test tests/integration/test_daemon_convergence_evidence.py -> 1 passed - devtools test tests/benchmarks/test_daemon_convergence.py tests/benchmarks/test_daemon_convergence_multi_provider.py -> 20 passed - devtools test tests/unit/devtools/test_benchmark_campaigns.py -> 10 passed - devtools verify --quick -> 16/16 steps ok (ruff format/check, mypy --strict, render all --check, layering, topology, closure-matrix, schema roundtrip, manifests, ci-workflows, doc-commands, docs-coverage, test-infra-currency, clock-hygiene, pytest-timeout-overrides, degrade-loudly) Ref polylogue-7uqr Co-Authored-By: Claude --- devtools/daemon_live_benchmark.py | 34 ++++++++----------- polylogue/daemon/cli.py | 17 +++------- polylogue/daemon/convergence.py | 16 +-------- tests/benchmarks/test_daemon_convergence.py | 4 +-- .../test_daemon_convergence_multi_provider.py | 2 +- .../test_daemon_convergence_evidence.py | 15 ++------ .../daemon/test_convergence_final_state.py | 5 +-- tests/unit/daemon/test_daemon_cli.py | 12 +------ tests/unit/daemon/test_standing_queries.py | 2 +- ...test_standing_queries_default_evaluator.py | 2 +- 10 files changed, 29 insertions(+), 80 deletions(-) diff --git a/devtools/daemon_live_benchmark.py b/devtools/daemon_live_benchmark.py index 18bff5e7c4..3d60e29426 100644 --- a/devtools/daemon_live_benchmark.py +++ b/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 = ( diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 56c9a5b2a3..26166b39f6 100644 --- a/polylogue/daemon/cli.py +++ b/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") diff --git a/polylogue/daemon/convergence.py b/polylogue/daemon/convergence.py index 64737effc5..e77833e988 100644 --- a/polylogue/daemon/convergence.py +++ b/polylogue/daemon/convergence.py @@ -166,15 +166,7 @@ class DaemonConverger: The main process is the only SQLite writer. """ - def __init__( - self, - stages: Iterable[ConvergenceStage], - *, - max_workers: int | None = None, - ) -> None: - # max_workers is accepted for caller compatibility but unused: no - # convergence stage runs in a worker process (polylogue-7uqr). - _ = max_workers + def __init__(self, stages: Iterable[ConvergenceStage]) -> None: self._stages: dict[str, ConvergenceStage] = {s.name: s for s in stages} self._file_states: dict[Path, FileState] = {} self._session_states: dict[str, SessionState] = {} @@ -271,12 +263,6 @@ def _session_barriers_blocked( return set(session_ids) return blocked.intersection(session_ids) - async def start(self) -> None: - logger.info("converger: started, stages=%s", list(self._stages)) - - async def stop(self) -> None: - 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: diff --git a/tests/benchmarks/test_daemon_convergence.py b/tests/benchmarks/test_daemon_convergence.py index 2aae4fb011..e821f56520 100644 --- a/tests/benchmarks/test_daemon_convergence.py +++ b/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), diff --git a/tests/benchmarks/test_daemon_convergence_multi_provider.py b/tests/benchmarks/test_daemon_convergence_multi_provider.py index dc6525a6ac..ecd5ce6107 100644 --- a/tests/benchmarks/test_daemon_convergence_multi_provider.py +++ b/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), diff --git a/tests/integration/test_daemon_convergence_evidence.py b/tests/integration/test_daemon_convergence_evidence.py index 83d973759b..c176103df3 100644 --- a/tests/integration/test_daemon_convergence_evidence.py +++ b/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, ( diff --git a/tests/unit/daemon/test_convergence_final_state.py b/tests/unit/daemon/test_convergence_final_state.py index 9f6e56c473..f3a50c3612 100644 --- a/tests/unit/daemon/test_convergence_final_state.py +++ b/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), diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py index 31a8672a16..9f73be76e4 100644 --- a/tests/unit/daemon/test_daemon_cli.py +++ b/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 diff --git a/tests/unit/daemon/test_standing_queries.py b/tests/unit/daemon/test_standing_queries.py index fe5a12e47a..0e893db395 100644 --- a/tests/unit/daemon/test_standing_queries.py +++ b/tests/unit/daemon/test_standing_queries.py @@ -204,7 +204,7 @@ def test_promoted_expected_count_divergence_targets_original_finding_without_wat mark_assertion_status(conn, original.assertion_id, AssertionStatus.ACCEPTED, now_ms=2) conn.commit() stage = make_standing_query_stage(index_db, evaluator=_Evaluator(members=("session:one", "session:two"))) - converger = DaemonConverger(stages=(stage,), max_workers=1) + converger = DaemonConverger(stages=(stage,)) states, _ = converger.converge_sessions(("session:changed",)) assert states["session:changed"].stages["standing-queries"].value == "done" diff --git a/tests/unit/daemon/test_standing_queries_default_evaluator.py b/tests/unit/daemon/test_standing_queries_default_evaluator.py index 456f58d1f2..4847bbf3fa 100644 --- a/tests/unit/daemon/test_standing_queries_default_evaluator.py +++ b/tests/unit/daemon/test_standing_queries_default_evaluator.py @@ -71,7 +71,7 @@ def test_default_stage_set_evaluates_a_watched_query_without_an_injected_fake(tm stages = make_default_convergence_stages(archive_root / "index.db") standing_stage = next(stage for stage in stages if stage.name == "standing-queries") - converger = DaemonConverger(stages=(standing_stage,), max_workers=1) + converger = DaemonConverger(stages=(standing_stage,)) states, _timings = converger.converge_sessions((session_id,)) assert states[session_id].stages["standing-queries"].value == "done"