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