From 7128b98acc2b23183a63a9190f9dbf5a21cb0cfe Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 20 Jul 2026 08:15:43 +0200 Subject: [PATCH] feat(daemon): suppress trickle conveyor while bulk rebuild is in flight Problem: PR #3189 (polylogue-gd6v) shipped daemon-owned bulk-scale index rebuild routing but left two residuals unaddressed: (1) the trickle raw-materialization conveyor and the bulk-rebuild driver both ran every tick, converging on the same backlog with duplicate parse/replay work and needless writer-hold contention, and (2) the p99 writer-responsiveness claim for a live drain was never independently measured. What changed: - polylogue/daemon/cli.py: new `_daemon_bulk_rebuild_transaction_in_flight` read-only check (mirrors `_maybe_route_daemon_bulk_rebuild`'s own `daemon_bulk_rebuild_routing` flag gate). `_periodic_raw_materialization_convergence` consults it at the top of its burst loop and, when a daemon bulk-rebuild transaction is already running, skips the trickle census/drain writer pass for that tick and drives the bulk pass instead, bursting at the same 1s cadence. The bulk engine's own paged cursor query (`IndexGenerationStore.next_raw_page`) is unfiltered by any fixed snapshot, so it naturally absorbs raws that arrive mid-build -- there is no work left for the trickle pass on the same raws while bulk is in-flight. `_maybe_route_daemon_bulk_rebuild` now returns a bool signal (pass genuinely attempted vs. structural no-op/swallowed failure) so a pass failure during suppression falls back to the slower outer interval instead of retrying every burst second. - Suppression scope is deliberately narrow: only the raw->index materialization work `_drain_raw_materialization_once` performs (the SAME work `docs/design/convergence-simplification-inventory.md` item 5 identifies as duplicated between the trickle and bulk engines). Separate periodic loops -- live-ingest acquisition, hook-spool drain, embedding catch-up, already-committed session insight convergence -- are untouched and keep running throughout a bulk drain. - tests/unit/daemon/test_daemon_cli.py: 6 new tests covering the in-flight check's flag gate, trickle suppression while in flight (the trickle drain must never run), failure fallback to the outer interval, and automatic resumption once the transaction is no longer in flight. Also backfilled `daemon_bulk_rebuild_routing = False` onto 4 pre-existing `FakeResolved` config stubs (parse-stage-split tests) that predate this attribute and broke once the new suppression check reads it unconditionally. - tests/unit/daemon/test_daemon_bulk_rebuild_responsiveness.py (new): drives the REAL `run_daemon_bulk_rebuild_pass` against a 150-raw fixture archive concurrently with 3 small writer-actor coroutines sharing the real `DaemonWriteCoordinator`, and asserts their p99 queued wait stays under 1.0s. Budget derived from manual measurement at this exact fixture shape: bounded passes (batch=8) measured p99 ~0.32s; collapsing the same corpus into one unbounded pass measured p99 ~1.26s -- the chosen budget sits between those two real numbers, so it is a genuine regression detector at this fixture size, not a loose ceiling. Verification: - devtools test tests/unit/daemon/test_daemon_cli.py tests/unit/daemon/test_bulk_rebuild.py tests/unit/daemon/test_daemon_bulk_rebuild_responsiveness.py tests/unit/daemon/test_write_coordinator.py tests/unit/daemon/test_parse_prefetch.py -> 134 passed - uv run mypy --strict on touched files -> clean - devtools verify --quick -> exit 0 (grepped for "out of sync", none) Ref polylogue-gd6v Co-Authored-By: Claude --- polylogue/daemon/cli.py | 88 +++++- ...test_daemon_bulk_rebuild_responsiveness.py | 260 ++++++++++++++++++ tests/unit/daemon/test_daemon_cli.py | 170 ++++++++++++ 3 files changed, 513 insertions(+), 5 deletions(-) create mode 100644 tests/unit/daemon/test_daemon_bulk_rebuild_responsiveness.py diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index d15b3a617a..2ec2e0ef3c 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -668,7 +668,37 @@ def _maybe_recommend_bulk_rebuild(counts: RawMaterializationCounts) -> None: ) -async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> None: +async def _daemon_bulk_rebuild_transaction_in_flight() -> bool: + """Whether the daemon's own well-known bulk-rebuild transaction is running. + + Read-only fast path (at most one JSON file read). Used by + ``_periodic_raw_materialization_convergence`` to stand its trickle + census/drain pass down for the current tick instead of duplicating work + the bulk-rebuild engine already subsumes: both walk the SAME + ``raw_sessions`` -> index materialization pipeline over the same backlog + (see docs/design/convergence-simplification-inventory.md item 5 -- the + bulk engine's own paged cursor query is unfiltered by any fixed + snapshot, so it naturally absorbs raws that arrive while it runs; there + is no work left for the trickle pass to do on the SAME raws in the + meantime). + + Mirrors ``_maybe_route_daemon_bulk_rebuild``'s own flag gate exactly: + the flag off means never even check. Checking survives a daemon restart + via a durable transaction record, so a stale "in flight" read while + routing is disabled would wrongly suppress the trickle conveyor -- the + ONLY mechanism left materializing raws -- with nothing to replace it. + """ + from polylogue.config import load_polylogue_config + + if not load_polylogue_config().daemon_bulk_rebuild_routing: + return False + from polylogue.daemon.bulk_rebuild import has_resumable_daemon_bulk_rebuild_transaction + from polylogue.paths import archive_root + + return await asyncio.to_thread(has_resumable_daemon_bulk_rebuild_transaction, archive_root()) + + +async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> bool: """polylogue-gd6v: route a bulk-scale backlog into a daemon-owned blue-green rebuild. Off by default (``daemon_bulk_rebuild_routing`` config flag). Once a @@ -679,11 +709,21 @@ async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> would waste every page already replayed into it. This runs after the trickle pass has already released the writer coordinator, so scheduling another writer-coordinated pass here is safe. + + Returns whether a pass was genuinely attempted this call (``True``) or + the call was a structural no-op / hit a swallowed failure (``False``). + ``_periodic_raw_materialization_convergence``'s trickle-suppression + branch (residual of polylogue-gd6v) uses this in place of the trickle + pass's own ``made_progress`` signal to decide whether to keep bursting + bulk-rebuild passes back-to-back or fall back to the slower outer + interval -- a genuine pass failure must not turn into a tight 1s retry + storm, matching how a trickle pass failure already falls back to the + outer interval via the caller's exception handling. """ from polylogue.config import load_polylogue_config if not load_polylogue_config().daemon_bulk_rebuild_routing: - return + return False from polylogue.config import Config from polylogue.daemon.bulk_rebuild import ( DAEMON_BULK_REBUILD_OPERATION_ID, @@ -696,7 +736,7 @@ async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> if not _bulk_scale_raw_materialization_backlog(counts) and not await asyncio.to_thread( has_resumable_daemon_bulk_rebuild_transaction, root ): - return + return False config = Config(archive_root=root, render_root=render_root(), sources=[]) try: receipt = await run_daemon_bulk_rebuild_pass( @@ -706,9 +746,9 @@ async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> ) except Exception: logger.warning("bulk-rebuild: routed pass failed", exc_info=True) - return + return False if receipt is None: - return + return True transaction_status = str(receipt.transaction["status"]) if receipt.transaction else receipt.status processed = receipt.transaction.get("processed_raw_count") if receipt.transaction else None logger.info( @@ -724,6 +764,7 @@ async def _maybe_route_daemon_bulk_rebuild(counts: RawMaterializationCounts) -> "the trickle conveyor's remaining backlog reflects the new active index from the next tick", DAEMON_BULK_REBUILD_OPERATION_ID, ) + return True async def _periodic_raw_materialization_convergence() -> None: @@ -745,6 +786,43 @@ async def _periodic_raw_materialization_convergence() -> None: recover = True try: while True: + # polylogue-gd6v residual: while the daemon's own bulk-rebuild + # transaction is in flight, it already subsumes exactly the + # raw->index materialization work this pass would do over + # the SAME backlog (see + # ``_daemon_bulk_rebuild_transaction_in_flight`` and + # docs/design/convergence-simplification-inventory.md item 5). + # Standing the trickle census/drain pass down here avoids + # both mechanisms converging on the same raws every tick -- + # double parse/replay work plus needless writer-hold + # contention between the two. Everything else the trickle + # conveyor is NOT responsible for (live-ingest acquisition, + # hook-spool drain, embedding catch-up, already-committed + # session insights) lives in separate periodic loops and + # keeps running unaffected. Driving the bulk pass here + # (instead of only from the trickle branch below) also lets + # bulk-rebuild progress burst at the same 1s cadence the + # trickle conveyor uses, rather than waiting a full quiet + # interval between passes. + if await _daemon_bulk_rebuild_transaction_in_flight(): + logger.debug( + "raw materialization: standing down trickle census/drain -- " + "a daemon bulk-rebuild transaction already subsumes this backlog" + ) + from polylogue.product.raw_authority import RawMaterializationCounts + + bulk_progressed = await _maybe_route_daemon_bulk_rebuild(RawMaterializationCounts()) + if not bulk_progressed: + # A swallowed pass failure -- fall back to the slower + # outer interval instead of retrying every burst + # second (mirrors how a trickle pass failure already + # escapes to the outer interval via the exception + # handlers below). + break + if _browser_capture_spool_has_pending_files(): + break + await asyncio.sleep(_RAW_MATERIALIZATION_BACKLOG_BURST_PAUSE_SECONDS) + continue # While replay planning is paused behind the persisted parser # census, a pass does census-only work: no replay transaction # runs, so the small replay-sized batch limit (which bounds diff --git a/tests/unit/daemon/test_daemon_bulk_rebuild_responsiveness.py b/tests/unit/daemon/test_daemon_bulk_rebuild_responsiveness.py new file mode 100644 index 0000000000..1e6843d3a5 --- /dev/null +++ b/tests/unit/daemon/test_daemon_bulk_rebuild_responsiveness.py @@ -0,0 +1,260 @@ +"""Fixture-scale responsiveness proof for polylogue-gd6v's remaining AC. + +PR #3189's own body deferred this explicitly: "agvo responsiveness p99 gate +during a live drain -- not independently measured here; rests on the same +off-writer-hold parse mechanism phase (a) already established." This module +supplies the missing measurement at fixture scale: it drives the REAL +``polylogue.daemon.bulk_rebuild.run_daemon_bulk_rebuild_pass`` -- the exact +production pass driver, not a stub -- against a real archive, CONCURRENTLY +with small simulated writer-actor coroutines (standing in for live-ingest +appends / hook-spool drain writes) sharing the SAME +``polylogue.daemon.write_coordinator.DaemonWriteCoordinator`` every other +daemon writer actor goes through, and asserts the small actors' queued-wait +time stays within a documented budget throughout the drain. + +Why this is a meaningful (non-vacuous) proof, not just a green assertion: + +* The coordinator is a strict FIFO single-writer gate (see + ``DaemonWriteCoordinator._execute``): once a small actor's request is + queued, its wait time is bounded by, at most, the currently-held pass's + remaining hold duration plus any earlier-queued items -- there is no + starvation-by-priority path. What actually determines whether that bound + is small is whether the *bulk-rebuild* side keeps its own passes bounded + (small ``raw_batch_size``, parse pre-warmed off the writer hold by + ``DaemonParseStage`` per #3168) instead of holding the writer for an + entire corpus in one sweep. +* Manual measurement during development, at this exact fixture shape + (150-raw corpus, batch=8, 3 concurrent small actors): bounded passes held + the writer for ~0.04-0.34s each and small-actor queued wait had p99 + ~0.32s. Collapsing the SAME corpus into one UNBOUNDED pass (batch=150, + the whole backlog in a single writer hold -- reproducing a regression + that removed per-pass batching, e.g. dropping ``RebuildIndexRequest``'s + paged ``raw_batch_size`` back to "whole backlog") measured a single + ~1.28s writer hold and pushed small-actor p99 wait to ~1.26s -- a small + actor queued behind that one giant hold waits for nearly the WHOLE + drain, not a bounded fraction of it. This test's budget (see + ``_SMALL_ACTOR_WAIT_BUDGET_SECONDS`` below) sits between those two + measurements: comfortably above the bounded-pass p99 (headroom against + host CPU contention -- this repo commonly runs concurrent agent/rebuild + load) while still well below what the unbounded-pass regression produces + at this exact fixture size, so a real regression to unbounded passes + would fail this test, not just a hypothetical one at a different scale. +""" + +from __future__ import annotations + +import asyncio +import json +import time +from pathlib import Path + +import pytest + +import polylogue.daemon.write_coordinator as write_coordinator_module +from polylogue.config import Config +from polylogue.core.enums import Provider +from polylogue.daemon.bulk_rebuild import run_daemon_bulk_rebuild_pass +from polylogue.daemon.parse_prefetch import DaemonParseStage +from polylogue.daemon.write_coordinator import DaemonWriteCoordinator, DaemonWriteEvent +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + +_RAW_COUNT = 150 +_BULK_BATCH_SIZE = 8 # forces >= 10 bounded passes over the fixture corpus +_SMALL_ACTOR_COUNT = 3 +_SMALL_ACTOR_INTERVAL_SECONDS = 0.02 +_MAX_PAYLOAD_BYTES = 10_000_000 + +# Manually measured at this exact fixture shape (150 raws, batch=8, 3 +# concurrent small actors): the correct bounded-pass implementation saw p99 +# queued wait ~0.32s (max single writer hold ~0.34s). Collapsing the SAME +# 150-raw corpus into one unbounded pass (batch=150) pushed p99 wait to +# ~1.26s. This budget sits between those two measurements: several times +# the bounded-pass p99 (headroom against host CPU contention -- this repo +# commonly runs concurrent agent/rebuild load) while staying below the +# unbounded-pass regression's measurement at this same fixture size, so +# this is a real (not merely hypothetical) regression detector, not just a +# loose ceiling nothing could ever hit. +_SMALL_ACTOR_WAIT_BUDGET_SECONDS = 1.0 + + +def _codex_session(native_id: str, messages: tuple[tuple[str, str], ...]) -> bytes: + rows: list[dict[str, object]] = [ + {"type": "session_meta", "payload": {"id": native_id, "timestamp": "2026-07-20T00:00:00Z"}} + ] + for position, (role, text) in enumerate(messages): + rows.append( + { + "type": "response_item", + "payload": { + "type": "message", + "id": f"{native_id}-m{position}", + "role": role, + "content": [ + {"type": "input_text" if role == "user" else "output_text", "text": text}, + ], + }, + } + ) + return b"".join(json.dumps(row, sort_keys=True).encode() + b"\n" for row in rows) + + +def _config(root: Path) -> Config: + return Config(archive_root=root, render_root=root / "render", sources=[]) + + +def _seed_corpus(root: Path, *, count: int = _RAW_COUNT) -> None: + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + for index in range(count): + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session( + f"responsiveness-session-{index}", + ( + ("user", f"question {index}"), + ("assistant", f"searchable answer {index}" * 10), + ), + ), + source_path=f"responsiveness-corpus-{index}.jsonl", + acquired_at_ms=index, + ) + + +def _small_write(_marker: int) -> None: + """Trivial fast writer-actor body -- stands in for a live-ingest append + or hook-spool drain write that must never queue for long behind a + bulk-rebuild pass sharing the same coordinator.""" + time.sleep(0.001) + + +async def _run_small_actor( + coordinator: DaemonWriteCoordinator, + name: str, + stop: asyncio.Event, + *, + interval: float, +) -> None: + counter = 0 + while not stop.is_set(): + await coordinator.run_sync(name, _small_write, counter) + counter += 1 + await asyncio.sleep(interval) + + +async def _drive_bulk_rebuild_to_promotion( + root: Path, + *, + batch_size: int, +) -> int: + """Drive the REAL daemon bulk-rebuild pass driver to promotion. + + Returns the number of bounded passes it took. Uses a fresh + ``DaemonParseStage`` per pass (mirroring a daemon restart between + ticks, same pattern as ``tests/unit/daemon/test_bulk_rebuild.py``) so + this also exercises the resume path rather than only a warm cache. + """ + config = _config(root) + pass_count = 0 + for _ in range(_RAW_COUNT * 2): # generous upper bound; promotion ends the loop early + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=_MAX_PAYLOAD_BYTES) + try: + receipt = await run_daemon_bulk_rebuild_pass( + config=config, + parse_stage=stage, + batch_size=batch_size, + max_payload_bytes=_MAX_PAYLOAD_BYTES, + ) + finally: + stage.shutdown() + if receipt is None: + break + pass_count += 1 + transaction_status = receipt.transaction["status"] if receipt.transaction else receipt.status + if transaction_status == "promoted": + break + else: + pytest.fail("bulk rebuild did not reach promotion within the generous pass budget") + return pass_count + + +def test_small_writer_actors_stay_responsive_during_bulk_rebuild_drain( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """gd6v residual: concurrent small writer actors must not be starved by + a real bulk-rebuild drain sharing the daemon write coordinator. + + Anti-vacuity: this drives ``run_daemon_bulk_rebuild_pass`` (the real + production pass driver used by ``_maybe_route_daemon_bulk_rebuild`` in + ``polylogue/daemon/cli.py``) against a real fixture archive, and the + small actors run through the real ``DaemonWriteCoordinator.run_sync`` -- + the exact same coordinator every other daemon writer actor (live + ingest, hook-spool drain, insight convergence) uses. A regression that + collapsed the bulk driver's own per-pass batching back into one + unbounded writer-held sweep (removing ``RebuildIndexRequest``'s paged + ``raw_batch_size``, or bypassing the coordinator's FIFO admission + entirely) would make at least one small actor wait for a hold + proportional to the WHOLE corpus instead of one bounded page, which + this fixture's corpus size (see module docstring) pushes well past + ``_SMALL_ACTOR_WAIT_BUDGET_SECONDS``. + """ + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(tmp_path)) + _seed_corpus(tmp_path) + + events: list[DaemonWriteEvent] = [] + coordinator = DaemonWriteCoordinator(observer=events.append) + # ``run_daemon_bulk_rebuild_pass`` resolves the coordinator via a local + # ``from polylogue.daemon.write_coordinator import daemon_write_coordinator`` + # import each call, so patching the module-level factory function makes + # every writer actor in this test -- the real bulk driver AND the small + # simulated actors below -- share this one instrumented instance, + # exactly like every writer actor in a real daemon process shares the + # one per-event-loop coordinator singleton. + monkeypatch.setattr(write_coordinator_module, "daemon_write_coordinator", lambda: coordinator) + + async def scenario() -> int: + stop = asyncio.Event() + small_actor_tasks = [ + asyncio.create_task( + _run_small_actor( + coordinator, + f"live.append.{i}", + stop, + interval=_SMALL_ACTOR_INTERVAL_SECONDS, + ) + ) + for i in range(_SMALL_ACTOR_COUNT) + ] + try: + return await _drive_bulk_rebuild_to_promotion(tmp_path, batch_size=_BULK_BATCH_SIZE) + finally: + stop.set() + for task in small_actor_tasks: + task.cancel() + await asyncio.gather(*small_actor_tasks, return_exceptions=True) + + pass_count = asyncio.run(scenario()) + + small_actor_waits = sorted( + event.wait_seconds + for event in events + if event.phase == "acquired" and event.actor.startswith("live.append.") and event.wait_seconds is not None + ) + bulk_pass_events = [ + event for event in events if event.phase == "acquired" and event.actor == "maintenance.bulk_rebuild" + ] + + # Sanity floor on the scenario itself: a single pass or a handful of + # small-actor samples would make the p99 assertion below vacuous (no + # real concurrency to interleave against). + assert pass_count >= 10, "fixture must force multiple bounded bulk passes to be a meaningful concurrency proof" + assert len(bulk_pass_events) == pass_count + assert len(small_actor_waits) >= 10, "small actors must genuinely interleave with the drain, not merely bookend it" + + p99_index = min(len(small_actor_waits) - 1, int(len(small_actor_waits) * 0.99)) + p99_wait = small_actor_waits[p99_index] + assert p99_wait < _SMALL_ACTOR_WAIT_BUDGET_SECONDS, ( + f"small writer actor p99 queued-wait {p99_wait:.3f}s exceeded the " + f"{_SMALL_ACTOR_WAIT_BUDGET_SECONDS}s budget while a real bulk-rebuild pass was draining " + f"({len(small_actor_waits)} samples across {pass_count} bulk passes)" + ) diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py index 1b486c3b49..38b2fd87ec 100644 --- a/tests/unit/daemon/test_daemon_cli.py +++ b/tests/unit/daemon/test_daemon_cli.py @@ -1271,6 +1271,7 @@ def test_periodic_raw_materialization_flag_off_never_warms_parse_stage( class FakeResolved: daemon_parse_stage_split = False + daemon_bulk_rebuild_routing = False seen_prefetch_cache: list[object] = [] @@ -1310,6 +1311,7 @@ def test_periodic_raw_materialization_flag_on_warms_off_writer_lease_before_drai class FakeResolved: daemon_parse_stage_split = True + daemon_bulk_rebuild_routing = False order: list[str] = [] lease_during_warm: list[bool] = [] @@ -1361,6 +1363,7 @@ def test_periodic_raw_materialization_flag_on_warm_exception_still_hands_back_ca class FakeResolved: daemon_parse_stage_split = True + daemon_bulk_rebuild_routing = False _sentinel_cache = object() seen_prefetch_cache: list[object] = [] @@ -1407,6 +1410,7 @@ def test_periodic_raw_materialization_flag_on_writer_hold_excludes_parse_stage_w class FakeResolved: daemon_parse_stage_split = True + daemon_bulk_rebuild_routing = False warm_delay_seconds = 0.2 released_hold_seconds: list[float] = [] @@ -4029,3 +4033,169 @@ async def fail_run_pass(**_kwargs: object) -> object: counts = RawMaterializationCounts(candidate_count=3, pending_blob_bytes=0) asyncio.run(daemon_cli._maybe_route_daemon_bulk_rebuild(counts)) # must not raise + + +def test_daemon_bulk_rebuild_transaction_in_flight_flag_off_never_checks( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The flag-off path must not even consult + ``has_resumable_daemon_bulk_rebuild_transaction`` -- mirrors + ``_maybe_route_daemon_bulk_rebuild``'s own flag gate exactly.""" + from polylogue.daemon import cli as daemon_cli + + class FakeResolved: + daemon_bulk_rebuild_routing = False + + def fail_has_resumable(_root: object) -> bool: + pytest.fail("must not check for a resumable transaction while the flag is off") + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr( + "polylogue.daemon.bulk_rebuild.has_resumable_daemon_bulk_rebuild_transaction", fail_has_resumable + ) + + assert asyncio.run(daemon_cli._daemon_bulk_rebuild_transaction_in_flight()) is False + + +def test_daemon_bulk_rebuild_transaction_in_flight_flag_on_delegates( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Flag on: delegates straight to ``has_resumable_daemon_bulk_rebuild_transaction`` + against the configured archive root.""" + from polylogue.daemon import cli as daemon_cli + + class FakeResolved: + daemon_bulk_rebuild_routing = True + + seen_roots: list[Path] = [] + + def fake_has_resumable(root: Path) -> bool: + seen_roots.append(root) + return True + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr( + "polylogue.daemon.bulk_rebuild.has_resumable_daemon_bulk_rebuild_transaction", fake_has_resumable + ) + + assert asyncio.run(daemon_cli._daemon_bulk_rebuild_transaction_in_flight()) is True + assert seen_roots == [tmp_path] + + +def test_periodic_raw_materialization_convergence_suppresses_trickle_while_bulk_rebuild_in_flight( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """polylogue-gd6v residual: the trickle census/drain pass must stand + down for the tick while the daemon's own bulk-rebuild transaction is in + flight, instead of both mechanisms converging on the same raw backlog + every tick (double parse/replay work + needless writer-hold contention). + """ + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + async def fake_in_flight() -> bool: + return True + + routed: list[RawMaterializationCounts] = [] + + async def fake_route(counts: RawMaterializationCounts) -> bool: + routed.append(counts) + return True + + async def fail_run_sync(actor: str, _func: object, *_args: object, **_kwargs: object) -> object: + pytest.fail(f"trickle drain must not run while a bulk-rebuild transaction is in flight (actor={actor})") + + async def fake_sleep(_seconds: float) -> None: + raise asyncio.CancelledError + + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_bulk_rebuild_transaction_in_flight", fake_in_flight) + monkeypatch.setattr(daemon_cli, "_maybe_route_daemon_bulk_rebuild", fake_route) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fail_run_sync)) + + with patch("asyncio.sleep", side_effect=fake_sleep), pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert len(routed) == 1 + assert routed[0].candidate_count == 0 # a placeholder counts object -- has_resumable alone gates routing + + +def test_periodic_raw_materialization_convergence_falls_back_to_outer_interval_on_bulk_rebuild_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A swallowed bulk-rebuild pass failure during suppression must not + turn into a tight 1s retry storm -- it falls back to the same slower + outer interval a trickle pass failure already falls back to.""" + from polylogue.daemon import cli as daemon_cli + + async def fake_in_flight() -> bool: + return True + + async def fake_route_fails(_counts: object) -> bool: + return False # mirrors _maybe_route_daemon_bulk_rebuild's own swallowed-failure return + + sleeps: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + raise asyncio.CancelledError + + async def fail_run_sync(actor: str, _func: object, *_args: object, **_kwargs: object) -> object: + pytest.fail(f"trickle drain must not run while a bulk-rebuild transaction is in flight (actor={actor})") + + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_bulk_rebuild_transaction_in_flight", fake_in_flight) + monkeypatch.setattr(daemon_cli, "_maybe_route_daemon_bulk_rebuild", fake_route_fails) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fail_run_sync)) + + with patch("asyncio.sleep", side_effect=fake_sleep), pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert sleeps == [daemon_cli._RAW_MATERIALIZATION_CONVERGENCE_INTERVAL_SECONDS] + + +def test_periodic_raw_materialization_convergence_resumes_trickle_once_bulk_rebuild_no_longer_in_flight( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Once ``has_resumable_daemon_bulk_rebuild_transaction`` flips false + (promoted or abandoned), the very next tick resumes the ordinary + trickle census/drain pass automatically -- no operator action, no wait + for the outer interval.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + in_flight_sequence = iter([True, False]) + + async def fake_in_flight() -> bool: + return next(in_flight_sequence) + + routed: list[object] = [] + + async def fake_route(counts: object) -> bool: + routed.append(counts) + return True + + trickle_calls: list[str] = [] + + async def fake_run_sync(actor: str, _func: object, *_args: object, **_kwargs: object) -> object: + trickle_calls.append(actor) + return RawMaterializationCounts(remaining_candidates=0) + + async def fake_sleep(seconds: float) -> None: + if seconds == daemon_cli._RAW_MATERIALIZATION_CONVERGENCE_INTERVAL_SECONDS: + raise asyncio.CancelledError + # burst-pause sleeps between suppressed passes: no-op, let the tick continue. + + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_bulk_rebuild_transaction_in_flight", fake_in_flight) + monkeypatch.setattr(daemon_cli, "_maybe_route_daemon_bulk_rebuild", fake_route) + monkeypatch.setattr(daemon_cli, "_maybe_recommend_bulk_rebuild", lambda _counts: None) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fake_run_sync)) + + with patch("asyncio.sleep", side_effect=fake_sleep), pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert trickle_calls == ["maintenance.raw_materialization"] + assert len(routed) >= 1 # the suppression branch drove at least one bulk pass first