diff --git a/devtools/docs_surface.py b/devtools/docs_surface.py index 2c342edb25..cde6487583 100644 --- a/devtools/docs_surface.py +++ b/devtools/docs_surface.py @@ -382,6 +382,12 @@ def _entry(title: str, path: str, description: str, tier: DocsTier) -> DocsEntry "Reference-blob representation for byte-proven superseded revision prefixes.", "design", ), + _entry( + "Convergence Simplification Inventory", + "design/convergence-simplification-inventory.md", + "Deletion/collapse inventory for the daemon convergence redesign (polylogue-m6tp).", + "design", + ), _entry("Second Brain", "design/second-brain.md", "Vision note for remembered work.", "design"), _entry("Time Machine", "design/time-machine.md", "Vision note for reconstructing work over time.", "design"), _entry("Whole Product", "design/whole-product.md", "Product vision and system relationships.", "design"), diff --git a/docs/README.md b/docs/README.md index 6287fe83f3..19c448791a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -117,6 +117,7 @@ Start with **Guides** for a task, **Reference** for a surface contract, and **Ar | [Session Lineage Model](design/session-lineage-model.md) | Fork, resume, compaction, and composition semantics. | | [Analysis Rigor](design/analysis-rigor.md) | Rigor mechanisms for agent claims: population validity and comparative judgment. | | [Prefix-Blob Reclamation](design/prefix-blob-reclamation.md) | Reference-blob representation for byte-proven superseded revision prefixes. | +| [Convergence Simplification Inventory](design/convergence-simplification-inventory.md) | Deletion/collapse inventory for the daemon convergence redesign (polylogue-m6tp). | | [Second Brain](design/second-brain.md) | Vision note for remembered work. | | [Time Machine](design/time-machine.md) | Vision note for reconstructing work over time. | | [Whole Product](design/whole-product.md) | Product vision and system relationships. | diff --git a/docs/configuration.md b/docs/configuration.md index 07a1200f6e..a3e29c506a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -349,6 +349,7 @@ A few keys not shown in the full example above, with their TOML path: | `ingest_commit_batch_messages` | `sources.ingest_commit_batch_messages` | Messages per commit batch during ingest (default 8000). | | `ingest_parse_workers` | `sources.ingest_parse_workers` | Parallel parse workers during ingest (default 1). | | `live_full_ingest_workers` | `sources.live_full_ingest_workers` | Parallel workers for a live full-reingest pass (default 1). | +| `daemon_parse_stage_split` | `daemon.raw_materialization.parse_stage_split` | Opt-in (polylogue-m6tp phase (a), default off): pre-parse raw-materialization census candidates in a bounded daemon-owned thread pool before the writer hold, instead of parsing inside the writer-held pass. | ## Environment Policy diff --git a/docs/design/README.md b/docs/design/README.md index 4201d8b4d0..3c3c1a1c1f 100644 --- a/docs/design/README.md +++ b/docs/design/README.md @@ -22,6 +22,7 @@ domain models rather than plans: | [Project memory](project-memory.md) · [Second brain](second-brain.md) · [Time machine](time-machine.md) · [Archive storytelling](archive-storytelling.md) · [Whole product](whole-product.md) | Vision statements feeding horizon beads | | [Query-action workflows](query-action-workflows.md) | Moved pointer to the generated `docs/product/workflows.md` | | [Prefix-blob reclamation](prefix-blob-reclamation.md) | Reference-blob representation for byte-proven superseded revision prefixes; consent-gated durable-tier reclamation (polylogue-vzn6) | +| [Convergence simplification inventory](convergence-simplification-inventory.md) | Deletion/collapse inventory for the daemon convergence redesign — what phases (b)-(d) remove and why (polylogue-m6tp) | If a doc here stops matching its owning beads, the beads win — update or purge the doc. diff --git a/docs/design/convergence-simplification-inventory.md b/docs/design/convergence-simplification-inventory.md new file mode 100644 index 0000000000..e17f21b1cb --- /dev/null +++ b/docs/design/convergence-simplification-inventory.md @@ -0,0 +1,296 @@ +# Convergence simplification inventory (polylogue-m6tp) + +Deletion/collapse inventory for the daemon convergence redesign (polylogue-m6tp, +related P0 polylogue-5jak). This document is deliberately scoped: it lists what +later phases of the redesign will delete or collapse, verified against the +current tree, and states why. **It deletes nothing itself** — phase (a) +(this PR) only adds the parse-stage extraction behind a config flag. Phases +(b)-(d) are tracked follow-up work on polylogue-m6tp. + +Read `docs/architecture.md`/`docs/internals.md` for the daemon's general +shape before reading this table; each row assumes the reader already knows +the census -> replay -> materialize pipeline. + +## Sequencing recap (from polylogue-m6tp's design sketch) + +1. **(a) parse-stage extraction behind a flag on the standard build** — this + PR. Proves the parse/apply seam works and is equivalence-safe; ships at + reduced benefit on a GIL build. +2. **(b) 3.14t (free-threaded) daemon deploy** — the same thread-pool code + path becomes a real 3.9x-9.6x parse speedup once the GIL is provably off + (`parallel_threads_effective()` gates this; see + `polylogue/pipeline/services/process_pool.py:62`). +3. **(c) bulk-scale routing** — candidate count/bytes above the + polylogue-m6tp threshold route to an in-process blue-green generation + build instead of the trickle conveyor. +4. **(d) deletions** — this table. Each mechanism below exists to work + around a constraint (process-pool spawn cost, GIL-era writer-starvation + risk, per-pass bounded-batch orchestration) that (b)/(c) remove. + +## Inventory + +### 1. Process-pool machinery + spawn workarounds + +**What it is:** `polylogue/pipeline/services/process_pool.py` — the shared +`ProcessPoolExecutor` helpers used by every CPU-bound parse dispatch on a +standard (GIL) build: + +- `process_pool_context()` (`polylogue/pipeline/services/process_pool.py:23`) + — forces the `spawn` start method specifically to avoid the forkserver + deadlock found in production (polylogue-p0pw: 17 minutes with zero parse + workers ever spawned, parent parked in `as_completed`). +- `process_pool_executor()` (`polylogue/pipeline/services/process_pool.py:93`) + — constructs a pool with `_initialize_worker_logging` as the per-worker + initializer, needed only because a spawned worker starts with a fresh, + unconfigured logging stack. +- `terminate_process_pool()` (`polylogue/pipeline/services/process_pool.py:102`) + — bounded-timeout cancel/terminate/kill sequence, needed only because a + process (unlike a thread) cannot be cooperatively interrupted from the + parent. +- `resolve_parse_worker_count()` (`polylogue/pipeline/services/process_pool.py:43`) + — resolves `POLYLOGUE_INGEST_PARSE_WORKERS` / cpu-1 default; the worker + *count* concept survives past this deletion (a thread pool still wants a + bound), only the process-specific plumbing goes. + +**Why it exists today:** on a standard CPython build, `ThreadPoolExecutor` +gives no CPU-bound parse speedup (the GIL serializes it) and, worse, running +parse threads concurrently with an actively write-holding thread measured +~5000x commit-latency inflation (the polylogue-7mtf control-run finding cited +throughout `revision_backfill.py`). `ProcessPoolExecutor` is the only way to +get real parallelism on this build, at the cost of spawn tax, pickling, and +no shared memory. + +**What makes it deletable:** phase (b)'s free-threaded 3.14t deploy makes a +plain `ThreadPoolExecutor` both safe (no writer-thread contention, since +phase (a) already sequences parse-then-apply so no writer thread is ever +active *during* parse) and fast (proven 3.9x-9.6x, zero writer interference +in the 7mtf control run). Once the daemon's runtime is provably free-threaded, +every process-pool call site collapses to the thread-pool call site that +phase (a) already introduces for the daemon's own conveyor +(`polylogue/daemon/parse_prefetch.py`) and that `revision_backfill.py` +already has for `parallel_threads_effective()`-gated callers +(`_parse_unique_retained_raws_via_threads`, +`polylogue/sources/revision_backfill.py:989`). + +**Which phase deletes it:** (b) removes the `ProcessPoolExecutor` branch from +every call site that currently gates on `parallel_threads_effective()` +(`polylogue/sources/revision_backfill.py:1066` `_parse_unique_retained_raws`); +`process_pool.py`'s process-specific helpers (`process_pool_context`, +`process_pool_executor`, `terminate_process_pool`) are deleted once no caller +remains. `resolve_parse_worker_count()`'s bound survives, retargeted at +thread-pool sizing. + +### 2. Pool-amortization heuristics (dispatch-size + aggregate-bytes floors) + +**What it is:** two independent guards in `polylogue/sources/revision_backfill.py` +that decide whether a batch is even worth spawning a process pool for: + +- `_partition_raws_by_dispatch_size()` (`polylogue/sources/revision_backfill.py:879`) + + `_parse_dispatch_max_bytes()` (`polylogue/sources/revision_backfill.py:857`, + default `_DEFAULT_PARSE_DISPATCH_MAX_BYTES = 262_144` / 256 KiB, override + `POLYLOGUE_REVISION_PARSE_DISPATCH_MAX_BYTES`) — raws at or above 256 KiB + parse sequentially in-process; the process-pool round trip pickles the + returned `ParsedSession` list back across the process boundary, which + measured a net LOSS (0.63x) above this size (polylogue-amg1/#3136). +- `_pool_dispatch_amortizes()` (`polylogue/sources/revision_backfill.py:922`) + + `_parse_pool_min_aggregate_bytes()` (`polylogue/sources/revision_backfill.py:899`, + default `_DEFAULT_PARSE_POOL_MIN_AGGREGATE_BYTES = 48 * 1024 * 1024` / 48 MiB, + override `POLYLOGUE_REVISION_PARSE_POOL_MIN_BYTES`) — an aggregate + pool-eligible batch under ~45 MB doesn't amortize the ~1.5-2s + per-worker spawn+import cost (measured live 2026-07-19: 20 short-lived + workers spending ~95% of their lifetime inside `importlib`). + +**Why it exists today:** both guards protect against process-pool-specific +costs (pickle-back of large payloads; per-worker spawn+import tax) that only +exist because `ProcessPoolExecutor` workers are separate interpreters. + +**What makes it deletable:** `_parse_unique_retained_raws_via_threads`'s own +docstring (`polylogue/sources/revision_backfill.py:989`) already states the +reason precisely: "Both `_partition_raws_by_dispatch_size` and +`_pool_dispatch_amortizes` exist solely to protect against those two +process-pool-specific costs (#3136/#3149), so this path applies NEITHER" — +a free-threaded `ThreadPoolExecutor` shares `ParsedSession` object graphs by +reference (no pickle) and reuses the one already-imported interpreter (no +per-worker spawn). Once the process-pool branch is gone (item 1), these two +size/aggregate floors have no remaining caller. + +**Which phase deletes it:** (b), in the same sweep as item 1 (they gate the +same dead branch). + +### 3. The 64 MiB daemon parse envelope narrowing + +**What it is:** `_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES = 64 * 1024 * 1024` +(`polylogue/daemon/cli.py:89`), threaded as `max_payload_bytes` into every +daemon-driven `repair_materialization` call +(`polylogue/daemon/cli.py:154` and `:850`). It caps how large a raw's blob +the daemon's conveyor will parse per pass; a raw above this envelope is +deferred (`record_resource_blocked_revision_census`, +`polylogue/sources/revision_backfill.py`) rather than parsed in-line, so one +whale raw cannot balloon the writer hold's memory footprint or duration. + +**Why it exists today:** parse currently happens *inside* the writer hold +(`daemon_write_coordinator().run_sync`, `polylogue/daemon/cli.py:685-692`), +so an unbounded parse of a multi-GB raw would hold the process-wide writer +lock — starving live ingest, status queries, and every other write actor — +for as long as that one parse takes. The 64 MiB ceiling is a blunt, +per-component admission gate that trades completeness (whales are refused, +not parsed) for a bounded worst-case hold duration. + +**What makes it deletable:** phase (a) (this PR) already moves the parse +itself off the writer hold via `DaemonParseStage` +(`polylogue/daemon/parse_prefetch.py`) — the memory/duration risk from a +large parse no longer threatens the writer hold's own duration once parse +runs before the hold is even requested. What replaces the blob-size ceiling +is `DaemonParseStage`'s explicit in-flight parsed-bytes budget +(`daemon_parse_stage_max_inflight_bytes()`, +`polylogue/daemon/parse_prefetch.py:72`, default 64 MiB, override +`POLYLOGUE_DAEMON_PARSE_STAGE_MAX_INFLIGHT_BYTES`) — a budget on cached +*parsed* memory, not a hard admission refusal on raw *blob* size. Once every +daemon parse path routes through the parse stage (phase (b)/(c) make this +the only path, not an opt-in flag), the per-component refusal ceiling +becomes redundant with the budget and can go. + +**Which phase deletes it:** (b)/(c) — specifically, once +`daemon_parse_stage_split` is no longer a flag (the parse-stage path is the +only path), `_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES` and its two call +sites in `polylogue/daemon/cli.py` are replaced by +`DaemonParseStage`'s budget alone. + +### 4. Census burst-escalation constants + +**What it is:** the daemon conveyor's bounded-pass sizing and back-to-back +burst logic in `polylogue/daemon/cli.py`: + +- `_RAW_MATERIALIZATION_CONVERGENCE_BATCH_LIMIT = 16` (`:80`) — replay-sized + per-pass limit (bounds writer-transaction length). +- `_RAW_MATERIALIZATION_CENSUS_BATCH_LIMIT = 64` (`:88`) — a larger + census-only-mode limit, because a census-paused pass runs no replay + transaction and the smaller replay-sized limit "only throttles + parse-bound census throughput and stretches a large census into days." +- `_RAW_MATERIALIZATION_BACKLOG_BURST_PAUSE_SECONDS = 1` (`:83`) — the yield + between back-to-back burst passes. +- The `census_mode` escalation switch itself + (`polylogue/daemon/cli.py:663`, `:678-681`, `:692-696`) — a pass that + censused components but repaired/executed nothing is treated as progress + and escalates the *next* pass's limit from 16 to 64. + +**Why it exists today:** this whole mechanism compensates for parse and +apply sharing one writer-held pass. Splitting "how many raws to census this +tick" from "how long the writer transaction can safely stay open" is the +root problem #3145/polylogue-m6tp/polylogue-5jak all describe — the batch +limit is doing double duty as both a parse-throughput knob and a +writer-hold-duration knob, and a single number cannot serve both jobs well +(too small starves census throughput on a census-paused backlog; too large +extends the writer hold on a replaying pass). + +**What makes it deletable:** once parse is a persistent, continuously-running +background stage (phase (b)/(c) make `DaemonParseStage` — or its bulk-mode +successor — an always-on backlog iterator rather than a per-pass batch), the +writer-held "apply" pass only needs to bound *its own* transaction length +(a much simpler, single-purpose number), and there is no separate +census-vs-replay batch-size distinction left to escalate between: census +throughput is bounded by the parse stage's own worker count and in-flight +budget, not by a per-tick candidate limit. + +**Which phase deletes it:** (b) collapses the census/replay batch-size +distinction once parse is continuously running rather than per-tick; +(c)'s persistent backlog iterator (item 5 below) removes the remaining +burst-pass bookkeeping (`census_mode`, the burst `while` loop's pause/yield +logic) entirely. + +### 5. Per-pass candidate requery / resume recompute + +**What it is:** `repair_raw_materialization` +(`polylogue/storage/repair.py:5695`) recomputes its FULL candidate set from +scratch via `_raw_materialization_candidate_ids()` up to twice per call: +once at entry (`polylogue/storage/repair.py:5783`) and again after the +census loop, to re-check what's still uncensused +(`polylogue/storage/repair.py:5844`). Each call re-scans `raw_sessions` +joined against `index_tier.sessions`/`raw_revision_applications`/ +`raw_membership_census` (`polylogue/storage/repair.py:3618` onward) — an +O(backlog size) query repeated every daemon tick regardless of how much of +the backlog actually changed since the previous tick. + +**Why it exists today:** the conveyor has no persistent memory of "where it +left off" beyond what's durably recorded in `source.db`/`index.db` +themselves (deliberately — restart-safety requires deriving the candidate +set from durable state, not an in-memory cursor that a crash would lose). +Recomputing from scratch is the simplest way to stay crash-consistent under +today's per-tick, stateless-between-ticks design. + +**What makes it deletable:** polylogue-m6tp's design sketch calls for "a +persistent in-daemon backlog iterator" that replaces per-pass candidate +requeries. Once the parse stage (and its eventual bulk-routing successor) +own a long-lived, incrementally-updated view of the pending backlog — fed by +the same durable receipts, but maintained incrementally rather than +recomputed by a full query every tick — a restart still recovers correctly +by rebuilding that view once at startup (not per tick), and steady-state +ticks no longer pay the full-backlog scan cost. + +**Which phase deletes it:** (c) — the persistent backlog iterator is +explicitly the mechanism the bulk-routing design introduces (needed there +regardless, to avoid re-scanning tens of thousands of raws once bulk-scale +generation building is in play); once it exists, the per-pass +`_raw_materialization_candidate_ids()` requery in +`repair_raw_materialization`'s trickle path becomes redundant with it. + +### 6. The CLI bulk importer's operator-surface status + +**What it is:** `polylogue ops maintenance rebuild-index` +(`polylogue/cli/commands/maintenance/_rebuild_index.py:281` +`@click.command("rebuild-index")`, handler `rebuild_index_command` at `:349`) +— today a live operator tool: #3145's daemon-side loud recommendation +(`polylogue/daemon/cli.py` `_maybe_recommend_bulk_rebuild`) tells an operator +to run it by hand when the trickle conveyor's backlog is bulk-scale, and the +2026-07-19 restore incident (polylogue-5jak notes) used it directly as the +only viable path once the daemon's own conveyor made a live backlog +net-negative. + +**Why it exists today:** it is the one code path that already does the +right thing for a bulk backlog — one resumable transaction, blue-green +generation, full parse envelope, one census+replay sweep — because it does +not share the daemon's live-ingest constraints (no concurrent watcher, no +per-tick writer-sharing budget, can run with the daemon stopped). + +**What makes it deletable as an *operator* tool (not as code):** polylogue-m6tp's +2026-07-19 operator direction states the target plainly: "with free-threaded +3.14t ... normal daemon convergence could BE the fast path, making the CLI +bulk importer unnecessary for ordinary backlogs." Phase (c)'s in-process +blue-green generation building (an inactive generation on a second writer +connection, live ingest continuing on the active index, promoted via the +existing generation-store pointer swap) gives the daemon itself everything +`rebuild-index` does today, without stopping the daemon or freezing the +source. Once that lands, an operator should never need to invoke +`rebuild-index` for routine backlog drains. + +**What survives:** the command itself is NOT deleted. Per the +automagic-invariants doctrine (an operator surface for routine work must be +subsumed by automatic convergence, not merely duplicated by it), it becomes +**disaster-recovery break-glass only** — the path used when the daemon +itself cannot run (corrupted state, daemon-dead scenarios; polylogue-k8kj's +robustness work already targets this exact scenario). Its resumable, +transactional, blue-green-generation design is exactly what a break-glass +path needs and should not be simplified away. + +**Which phase deletes/collapses it:** (c) is what makes the daemon's own +convergence loop sufficient for ordinary bulk backlogs; the CLI command's +*operator-tool* status (documented recommendation, routine-use expectation) +is retired at that point, while the command and its underlying +`polylogue/maintenance/rebuild_index.py` machinery remain as the +break-glass path. No code deletion is scoped here — only a documentation/ +recommendation change (stop telling operators to run it routinely) plus, +optionally, gating it behind an explicit `--i-know-the-daemon-is-down` style +confirmation in a later bead if the break-glass framing needs to be load-bearing +in the CLI itself. + +## What phase (a) (this PR) does NOT touch + +For clarity, since this document sits next to the parse-stage extraction +PR: none of the six items above are deleted, narrowed, or behaviorally +changed by phase (a). Every mechanism above continues to run exactly as +before when `daemon_parse_stage_split` is off (the default), and continues +to run unchanged even when the flag is on for every code path except the +one new prefetch-cache-hit shortcut in `_parse_retained_raws` +(`polylogue/sources/revision_backfill.py`), which is additive and +equivalence-tested (see +`tests/unit/daemon/test_raw_materialization_parse_stage_equivalence.py`). diff --git a/docs/plans/test-clock-allowlist.yaml b/docs/plans/test-clock-allowlist.yaml index c6d0a8cfb6..926ce1b4bf 100644 --- a/docs/plans/test-clock-allowlist.yaml +++ b/docs/plans/test-clock-allowlist.yaml @@ -75,6 +75,8 @@ files: reason: "Metrics endpoint test records a live rebuild-ingest heartbeat timestamp; production readiness and metrics intentionally compare it against the host clock." - path: tests/unit/daemon/test_daemon_lifecycle.py reason: "Lifecycle signal-forensics test holds a real SQLite write lock and measures that the terminating handler respects its bounded busy timeout." + - path: tests/unit/daemon/test_parse_prefetch.py + reason: "DaemonParseStage.warm() timeout test measures real elapsed wall-clock against a genuinely hung worker thread to prove the wait is bounded, not merely reordered; frozen_clock cannot substitute for a real ThreadPoolExecutor future's wait timeout." - path: tests/integration/test_daemon_resilience.py reason: "Daemon resilience integration tests (#1735) measure real elapsed wall-clock for process lifecycle events (SIGKILL delivery, subprocess startup, concurrency timing). frozen_clock cannot substitute for real time when waiting on OS process state." - path: tests/integration/test_ingest_pipeline_correctness.py diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index e1d73568ed..9027e4f991 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -1258,7 +1258,7 @@ files: target: polylogue/cli/verb_names.py owner: stable - path: polylogue/config.py - loc: 2304 + loc: 2329 target: polylogue/config.py owner: kernel reason: kernel root rule @@ -1484,7 +1484,7 @@ files: target: polylogue/daemon/catchup_status.py owner: stable - path: polylogue/daemon/cli.py - loc: 2334 + loc: 2403 target: polylogue/daemon/cli.py owner: stable - path: polylogue/daemon/compare.py @@ -1628,6 +1628,10 @@ files: loc: 288 target: polylogue/daemon/otlp_receiver.py owner: stable + - path: polylogue/daemon/parse_prefetch.py + loc: 179 + target: polylogue/daemon/parse_prefetch.py + owner: stable - path: polylogue/daemon/process_start.py loc: 23 target: polylogue/daemon/process_start.py @@ -2075,7 +2079,7 @@ files: target: polylogue/maintenance/preview.py owner: stable - path: polylogue/maintenance/rebuild_index.py - loc: 413 + loc: 496 target: polylogue/maintenance/rebuild_index.py owner: stable - path: polylogue/maintenance/registry.py @@ -2083,7 +2087,7 @@ files: target: polylogue/maintenance/registry.py owner: stable - path: polylogue/maintenance/replay.py - loc: 826 + loc: 835 target: polylogue/maintenance/replay.py owner: stable - path: polylogue/maintenance/scope.py @@ -2433,7 +2437,7 @@ files: target: polylogue/product/continuity_scenarios.py owner: stable - path: polylogue/product/raw_authority.py - loc: 126 + loc: 138 target: polylogue/product/raw_authority.py owner: stable - path: polylogue/product/workflows.py @@ -3032,7 +3036,7 @@ files: target: polylogue/sinex/transport.py owner: stable - path: polylogue/sources/__init__.py - loc: 71 + loc: 72 target: polylogue/sources/__init__.py owner: stable - path: polylogue/sources/assembly.py @@ -3357,7 +3361,7 @@ files: owner: stable cross_cut: { lifecycle: model } - path: polylogue/sources/revision_backfill.py - loc: 1213 + loc: 1342 target: polylogue/sources/revision_backfill.py owner: stable - path: polylogue/sources/source_acquisition.py @@ -3542,7 +3546,7 @@ files: target: polylogue/storage/fts/freshness.py owner: stable - path: polylogue/storage/fts/fts_lifecycle.py - loc: 1031 + loc: 1051 target: polylogue/storage/fts/fts_lifecycle.py owner: stable - path: polylogue/storage/fts/pl_fold.py @@ -3554,7 +3558,7 @@ files: target: polylogue/storage/fts/session_repair.py owner: stable - path: polylogue/storage/fts/sql.py - loc: 225 + loc: 252 target: polylogue/storage/fts/sql.py owner: stable - path: polylogue/storage/hydrators.py @@ -3568,7 +3572,7 @@ files: owner: storage-root reason: storage-root cross-cutting helper - path: polylogue/storage/index_generation.py - loc: 530 + loc: 544 target: TBD owner: storage-domain - path: polylogue/storage/insights/__init__.py @@ -3705,7 +3709,7 @@ files: owner: storage-root reason: storage-root cross-cutting helper - path: polylogue/storage/repair.py - loc: 6532 + loc: 6640 target: polylogue/storage/repair.py owner: storage-root reason: storage-root cross-cutting helper @@ -3913,7 +3917,7 @@ files: target: polylogue/storage/sqlite/__init__.py owner: stable - path: polylogue/storage/sqlite/action_pairs.py - loc: 72 + loc: 145 target: polylogue/storage/sqlite/action_pairs.py owner: stable - path: polylogue/storage/sqlite/action_relation.py @@ -3925,7 +3929,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/__init__.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive.py - loc: 12419 + loc: 12438 target: polylogue/storage/sqlite/archive_tiers/archive.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive_init.py @@ -3965,7 +3969,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/embeddings.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index.py - loc: 1636 + loc: 1649 target: polylogue/storage/sqlite/archive_tiers/index.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/ingest_precedence.py @@ -3985,7 +3989,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/pricing_seed.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/revision_application.py - loc: 298 + loc: 313 target: polylogue/storage/sqlite/archive_tiers/revision_application.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/self_verify.py @@ -4029,7 +4033,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/user_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/write.py - loc: 4809 + loc: 4908 target: polylogue/storage/sqlite/archive_tiers/write.py owner: stable - path: polylogue/storage/sqlite/async_sqlite.py @@ -4053,7 +4057,7 @@ files: target: polylogue/storage/sqlite/connection_profile.py owner: stable - path: polylogue/storage/sqlite/delegation_facts.py - loc: 60 + loc: 87 target: polylogue/storage/sqlite/delegation_facts.py owner: stable - path: polylogue/storage/sqlite/finding_provenance.py @@ -4065,7 +4069,7 @@ files: target: polylogue/storage/sqlite/holdout_cohorts.py owner: stable - path: polylogue/storage/sqlite/lifecycle.py - loc: 318 + loc: 334 target: polylogue/storage/sqlite/lifecycle.py owner: stable - path: polylogue/storage/sqlite/maintenance.py diff --git a/docs/topology-status.md b/docs/topology-status.md index 55bc378372..7d27a856f7 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 882 +- **Stable** (no move scoped): 883 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 9 -- **Total declared**: 1053 -- **Realized polylogue/**/*.py**: 1053 files declared +- **Total declared**: 1054 +- **Realized polylogue/**/*.py**: 1054 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/config.py b/polylogue/config.py index 1f41641de4..79bb64da3b 100644 --- a/polylogue/config.py +++ b/polylogue/config.py @@ -553,6 +553,15 @@ def ingest_parse_workers(self) -> int: def live_full_ingest_workers(self) -> int: return max(1, int(str(self._data.get("live_full_ingest_workers", 1)))) + @property + def daemon_parse_stage_split(self) -> bool: + """Opt-in: pre-parse raw-materialization census candidates off the writer hold. + + polylogue-m6tp phase (a). Off by default. See + ``polylogue.daemon.parse_prefetch.DaemonParseStage``. + """ + return bool(self._data.get("daemon_parse_stage_split")) + def get(self, key: str, default: object = None) -> object: value = self._data.get(key, default) return _thaw_config_value(value) @@ -1125,6 +1134,20 @@ def effective_path(self) -> str: description="Subscription plan rows used by cost/outlook reporting.", toml_kind="array-table", ), + ConfigInventoryEntry( + "daemon_parse_stage_split", + toml_path="daemon.raw_materialization.parse_stage_split", + env_var="POLYLOGUE_DAEMON_PARSE_STAGE_SPLIT", + owner_class="resource-policy", + reload_behavior="daemon-loop", + description=( + "Opt-in (polylogue-m6tp phase (a)): pre-parse raw-materialization " + "census candidates in a bounded daemon-owned thread pool BEFORE " + "the writer hold, instead of parsing inside the writer-held pass. " + "Off by default; proves the parse/apply seam ahead of the " + "free-threaded 3.14t daemon deploy." + ), + ), ) _CONFIG_INVENTORY_BY_KEY = {entry.key: entry for entry in _CONFIG_INVENTORY} @@ -1158,6 +1181,7 @@ def effective_path(self) -> str: "notification_email_use_tls", "notification_email_use_starttls", "observability_enabled", + "daemon_parse_stage_split", } ) @@ -1348,6 +1372,7 @@ def _default_config_values(bootstrap: _BootstrapPaths | None = None) -> dict[str "ingest_parse_workers": default_parse_workers, "live_full_ingest_workers": 1, "subscription_plans": (), + "daemon_parse_stage_split": False, } diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index bc5b30e504..56c9a5b2a3 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -67,7 +67,9 @@ if TYPE_CHECKING: from polylogue.daemon.lifecycle import DaemonLifecycle + from polylogue.daemon.parse_prefetch import DaemonParseStage from polylogue.product.raw_authority import RawMaterializationCounts + from polylogue.sources.revision_backfill import RawParsePrefetchCache logger = get_logger(__name__) _CONVERGENCE_DEBT_RETRY_INTERVAL_SECONDS = 60 @@ -104,6 +106,60 @@ _BULK_REBUILD_RECOMMENDATION_MIN_INTERVAL_SECONDS = 3600.0 # at most once/hour _last_bulk_rebuild_recommendation_monotonic: float | None = None +# polylogue-m6tp phase (a): one parse-stage warmer lives for the daemon +# process's lifetime, lazily created the first time the +# ``daemon_parse_stage_split`` config flag is observed on. It is deliberately +# module-level (not per-pass) so its bounded ``ThreadPoolExecutor`` and +# ``RawParsePrefetchCache`` persist across ticks -- a raw warmed but not +# consumed this pass (component-grouping selects a different subset than the +# flat candidate preview) remains cached for a later one. +_daemon_parse_stage_singleton: DaemonParseStage | None = None + + +def _daemon_parse_stage() -> DaemonParseStage: + global _daemon_parse_stage_singleton + if _daemon_parse_stage_singleton is None: + from polylogue.daemon.parse_prefetch import DaemonParseStage + + _daemon_parse_stage_singleton = DaemonParseStage() + return _daemon_parse_stage_singleton + + +async def _maybe_warm_raw_materialization_parse_stage(*, limit: int) -> RawParsePrefetchCache | None: + """Pre-parse this pass's census candidates outside the writer hold. + + polylogue-m6tp phase (a). Off by default (``daemon_parse_stage_split`` + config flag); returns ``None`` unless enabled, which makes + ``_drain_raw_materialization_once`` parse every candidate inside the + writer hold exactly as before -- the unmodified, always-correct + behavior. Runs entirely BEFORE the write coordinator is ever asked for + the writer hold, so it never competes with an active writer thread for + the GIL (see ``polylogue.daemon.parse_prefetch`` for why that sequencing + is what makes threads safe here even on a standard GIL build). + """ + from polylogue.config import load_polylogue_config + + if not load_polylogue_config().daemon_parse_stage_split: + return None + from polylogue.config import Config + from polylogue.paths import archive_root, render_root + + stage = _daemon_parse_stage() + config = Config(archive_root=archive_root(), render_root=render_root(), sources=[]) + try: + warmed = await asyncio.to_thread( + stage.warm, + config, + limit=limit, + max_payload_bytes=_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES, + ) + except Exception: + logger.warning("raw materialization: parse-stage prefetch failed; falling back to in-hold parse", exc_info=True) + return stage.cache + if warmed: + logger.info("raw materialization: parse-stage prefetch warmed %d raw(s) off the writer hold", warmed) + return stage.cache + async def _run_startup_fts_readiness(coordinator: DaemonWriteCoordinator) -> None: """Run the real startup FTS writer on an exit-safe coordinator thread.""" @@ -625,9 +681,15 @@ async def _periodic_raw_materialization_convergence() -> None: if census_mode else _RAW_MATERIALIZATION_CONVERGENCE_BATCH_LIMIT ) + prefetch_cache = await _maybe_warm_raw_materialization_parse_stage(limit=limit) materialized = await daemon_write_coordinator().run_sync( "maintenance.raw_materialization", - functools.partial(_drain_raw_materialization_once, limit=limit, recover=recover), + functools.partial( + _drain_raw_materialization_once, + limit=limit, + recover=recover, + prefetch_cache=prefetch_cache, + ), ) recover = False census_mode = ( @@ -742,12 +804,18 @@ def _drain_raw_materialization_once( *, limit: int = _RAW_MATERIALIZATION_CONVERGENCE_BATCH_LIMIT, recover: bool = True, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> Any: """Run one bounded raw source→index convergence pass. ``recover`` gates the interrupted-frontier recovery scan: it only has work after a crash/restart, so backlog burst continuations within one healthy cycle skip it instead of re-scanning per pass. + + ``prefetch_cache`` (polylogue-m6tp phase (a), default ``None``) is + populated by ``_maybe_warm_raw_materialization_parse_stage`` BEFORE this + function is ever scheduled onto the writer hold; passing ``None`` + (the flag-off default) reproduces the exact unmodified in-hold parse. """ from polylogue.config import Config from polylogue.paths import archive_root, render_root @@ -780,6 +848,7 @@ def _drain_raw_materialization_once( dry_run=False, raw_artifact_limit=limit, max_payload_bytes=_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES, + prefetch_cache=prefetch_cache, ) finally: _close_raw_materialization_fts(config.archive_root / "index.db") @@ -1758,6 +1827,16 @@ async def run_daemon_services( 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. + _daemon_parse_stage_singleton.shutdown() if server is not None: await _shutdown_server_if_serving(server, server_task, label="browser-capture") if api_server is not None: diff --git a/polylogue/daemon/parse_prefetch.py b/polylogue/daemon/parse_prefetch.py new file mode 100644 index 0000000000..d2f27eaba5 --- /dev/null +++ b/polylogue/daemon/parse_prefetch.py @@ -0,0 +1,240 @@ +"""Daemon-owned parse-stage extraction: parse census candidates off the writer hold. + +polylogue-m6tp phase (a). The raw-materialization conveyor's writer hold +(``DaemonWriteCoordinator.run_sync``) used to cover BOTH the CPU-bound +blob->``ParsedSession`` decode (census parse) and the SQLite writes that +record it, so a large or slow parse extended the writer hold by exactly as +long as the parse took -- starving every other write actor (live ingest, +status snapshots, insight convergence) queued behind the same coordinator. + +This module lets the daemon pre-parse the NEXT pass's candidate raws in a +bounded ``ThreadPoolExecutor`` BEFORE the writer hold is ever requested. The +writer-held pass then finds those results already warmed in a +``RawParsePrefetchCache`` (``polylogue.sources.revision_backfill``) and skips +reparsing them -- see that class's docstring for why a miss (empty cache, +budget-rejected entry, or the flag simply being off) always degrades to the +exact unmodified parse path rather than incorrect behavior. + +Why threads are safe here even on a standard (GIL) build: the polylogue-7mtf +control-run measurement (``parallel_threads_effective`` in +``polylogue.pipeline.services.process_pool``) found threaded parse gives no +GIL-build speedup AND inflates a *concurrently write-holding* thread's commit +latency ~5000x. That hazard is specifically about a parse thread running +WHILE a writer thread is active. This module never does that: ``warm()`` is +called by the conveyor BEFORE it ever asks the write coordinator for the +writer hold, so there is no writer thread to contend with. On a GIL build +this still gives little or no wall-clock parse speedup (CPython serializes +the CPU-bound decode across threads) -- that is expected and is the point of +phase (a): prove the parse/apply seam is correct and equivalence-safe ahead +of the free-threaded 3.14t deploy (phase (b), polylogue-m6tp), which is what +turns the same code path into a real speedup. +""" + +from __future__ import annotations + +import os +from concurrent.futures import ThreadPoolExecutor, as_completed + +from polylogue.config import Config +from polylogue.logging import get_logger +from polylogue.sources import revision_backfill +from polylogue.sources.dispatch import is_stream_record_provider +from polylogue.sources.revision_backfill import RawParsePrefetchCache +from polylogue.storage.repair import ( + raw_materialization_pending_census_raw_ids, + raw_materialization_readonly_descriptors, +) + +logger = get_logger(__name__) + +_DEFAULT_MAX_INFLIGHT_BYTES = 64 * 1024 * 1024 # 64 MiB + +# CodeRabbit (PR #3168): as_completed()/future.result() had no timeout, so one +# hung worker (e.g. an unresponsive filesystem read) would block warm() +# forever -- and warm() is awaited directly ahead of run_sync in the periodic +# raw-materialization loop, so a stuck warm pass would stall every subsequent +# drain pass indefinitely, not just this one. 300s (5 min) is generous for +# the happy path (a bounded batch of already-published local blob reads) and +# only ever matters on a genuine hang. On timeout, still-pending raws are +# simply left uncached -- the writer-held pass reparses them normally, the +# same graceful-degradation guarantee as any other prefetch miss. A +# ThreadPoolExecutor cannot forcibly kill a running worker thread, so a truly +# wedged worker keeps occupying one pool slot until it (eventually) returns; +# that is an inherent limitation of thread-based cancellation, not something +# this bound can fix -- the bound's job is only to stop the CONVEYOR LOOP +# from waiting on it forever, which it does. +_DEFAULT_WARM_TIMEOUT_SECONDS = 300.0 + + +def daemon_parse_stage_worker_count() -> int: + """Bounded worker cap for the daemon-owned pre-parse thread pool. + + ``cpu_count - 1`` leaves one core free for the daemon's own event loop, + mirroring ``resolve_parse_worker_count``'s cpu-1 convention (see + ``polylogue.pipeline.services.process_pool``). Override with + ``POLYLOGUE_DAEMON_PARSE_STAGE_WORKERS``. + """ + raw = os.environ.get("POLYLOGUE_DAEMON_PARSE_STAGE_WORKERS") + if raw is not None: + try: + value = int(raw) + except ValueError: + value = 0 + if value > 0: + return value + return max(1, (os.cpu_count() or 2) - 1) + + +def daemon_parse_stage_max_inflight_bytes() -> int: + """Whale-memory budget for parsed sessions held in the prefetch cache. + + Override with ``POLYLOGUE_DAEMON_PARSE_STAGE_MAX_INFLIGHT_BYTES``. + """ + raw = os.environ.get("POLYLOGUE_DAEMON_PARSE_STAGE_MAX_INFLIGHT_BYTES") + if raw is not None: + try: + value = int(raw) + except ValueError: + value = 0 + if value > 0: + return value + return _DEFAULT_MAX_INFLIGHT_BYTES + + +def daemon_parse_stage_warm_timeout_seconds() -> float: + """Bound on how long ``warm()`` waits for its dispatched workers. + + Override with ``POLYLOGUE_DAEMON_PARSE_STAGE_WARM_TIMEOUT_SECONDS``. See + ``_DEFAULT_WARM_TIMEOUT_SECONDS`` for why this exists and what it does + (and does not) guarantee. + """ + raw = os.environ.get("POLYLOGUE_DAEMON_PARSE_STAGE_WARM_TIMEOUT_SECONDS") + if raw is not None: + try: + value = float(raw) + except ValueError: + value = 0.0 + if value > 0: + return value + return _DEFAULT_WARM_TIMEOUT_SECONDS + + +class DaemonParseStage: + """Owns the daemon's bounded pre-parse ``ThreadPoolExecutor`` and cache. + + One instance lives for the daemon process's lifetime (created lazily by + the raw-materialization conveyor loop when ``daemon_parse_stage_split`` + is enabled). ``warm`` is synchronous/blocking -- callers run it off the + event loop (``asyncio.to_thread``), exactly like every other conveyor + pass, and NEVER under ``daemon_write_coordinator().run_sync``: doing so + would defeat the entire point, since the pre-parse must run without the + writer hold held. + """ + + def __init__( + self, + *, + max_workers: int | None = None, + max_inflight_bytes: int | None = None, + warm_timeout_seconds: float | None = None, + ) -> None: + self._executor = ThreadPoolExecutor( + max_workers=max_workers if max_workers is not None else daemon_parse_stage_worker_count(), + thread_name_prefix="polylogue-parse-stage", + ) + self.cache = RawParsePrefetchCache( + max_inflight_bytes=( + max_inflight_bytes if max_inflight_bytes is not None else daemon_parse_stage_max_inflight_bytes() + ) + ) + self._warm_timeout_seconds = ( + warm_timeout_seconds if warm_timeout_seconds is not None else daemon_parse_stage_warm_timeout_seconds() + ) + + def warm(self, config: Config, *, limit: int, max_payload_bytes: int) -> int: + """Pre-parse up to ``limit`` pending census candidates outside any writer hold. + + Returns the number of raws newly admitted to the cache. Read-only + end to end: candidate discovery and descriptor lookup both open + ``mode=ro`` SQLite connections (``polylogue.storage.repair``); + parsing reads only already-published blob bytes via a stateless + ``ArchiveBlobPublisher``, mirroring the production census parse + worker exactly (``census_parse_worker``, the same function the + writer-held path dispatches to a process/thread pool). Nothing here + writes to source.db, index.db, or takes the daemon's writer lease. + """ + candidate_raw_ids = raw_materialization_pending_census_raw_ids( + config, limit=limit, max_payload_bytes=max_payload_bytes + ) + raw_ids = [raw_id for raw_id in candidate_raw_ids if not self.cache.contains(raw_id)] + if not raw_ids: + return 0 + archive_root = config.archive_root + descriptors = raw_materialization_readonly_descriptors(archive_root, raw_ids) + blob_root_str = str(archive_root / "blob") + source_db_path_str = str(archive_root / "source.db") + + futures = {} + for raw_id in raw_ids: + descriptor = descriptors.get(raw_id) + if descriptor is None: + continue + provider, blob_hash, source_path, _kind, _size = descriptor + future = self._executor.submit( + revision_backfill.census_parse_worker, + raw_id, + provider.value, + blob_hash, + source_path, + is_stream_record_provider(source_path, str(provider)), + blob_root_str, + source_db_path_str, + ) + futures[future] = raw_id + + warmed = 0 + completed = 0 + try: + for future in as_completed(futures, timeout=self._warm_timeout_seconds): + completed += 1 + raw_id = futures[future] + try: + _raw_id, sessions, error = future.result() + except Exception: + logger.warning("parse-stage prefetch: worker failed for raw_id=%s", raw_id, exc_info=True) + continue + if error is not None or sessions is None: + # Parse failures are intentionally NOT cached: the writer-held + # pass reparses (and correctly quarantines/records) this raw + # exactly as it would with the flag off. Prefetch only ever + # shortcuts the happy path. + continue + _provider, _blob_hash, _source_path, kind, payload_size = descriptors[raw_id] + if self.cache.try_admit(raw_id, sessions, payload_bytes=payload_size, revision_kind=kind): + warmed += 1 + except TimeoutError: + # Bounds the CONVEYOR LOOP's wait, not the worker itself -- a + # ThreadPoolExecutor cannot forcibly kill a running thread, so a + # genuinely wedged worker keeps occupying one pool slot until it + # eventually returns (see _DEFAULT_WARM_TIMEOUT_SECONDS). Every raw + # not yet completed is simply left uncached: the writer-held pass + # reparses it normally, identical to any other prefetch miss. + pending = len(futures) - completed + logger.warning( + "parse-stage prefetch: warm() timed out after %.0fs waiting on %d of %d worker(s); " + "leaving unfinished raw(s) uncached for the writer-held pass to reparse normally", + self._warm_timeout_seconds, + pending, + len(futures), + ) + return warmed + + def shutdown(self) -> None: + self._executor.shutdown(wait=False, cancel_futures=True) + + +__all__ = [ + "DaemonParseStage", + "daemon_parse_stage_max_inflight_bytes", + "daemon_parse_stage_worker_count", +] diff --git a/polylogue/product/raw_authority.py b/polylogue/product/raw_authority.py index 48c1086e65..8315d0bd1b 100644 --- a/polylogue/product/raw_authority.py +++ b/polylogue/product/raw_authority.py @@ -9,11 +9,14 @@ from dataclasses import dataclass from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any from polylogue.config import Config from polylogue.core.json import JSONDocument +if TYPE_CHECKING: + from polylogue.sources.revision_backfill import RawParsePrefetchCache + @dataclass(frozen=True, slots=True) class RawMaterializationCounts: @@ -72,7 +75,15 @@ def repair_materialization( dry_run: bool, raw_artifact_limit: int, max_payload_bytes: int, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> Any: + """Run one bounded raw source->index convergence pass. + + ``prefetch_cache`` (polylogue-m6tp phase (a), default ``None``) lets a + caller substitute parse output already computed off the writer hold for + this pass's census phase; see + ``polylogue.sources.revision_backfill.RawParsePrefetchCache``. + """ from polylogue.storage.repair import repair_raw_materialization return repair_raw_materialization( @@ -80,6 +91,7 @@ def repair_materialization( dry_run=dry_run, raw_artifact_limit=raw_artifact_limit, max_payload_bytes=max_payload_bytes, + prefetch_cache=prefetch_cache, ) diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index bcd58ec884..bac3dfce90 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -7,6 +7,7 @@ import pickle import sqlite3 import tempfile +import threading from collections.abc import Callable, Iterator from contextlib import contextmanager from dataclasses import dataclass @@ -61,6 +62,81 @@ class _RevisionCensusState: provisional_full_raw_ids: dict[str, set[str]] +@dataclass(slots=True) +class _PrefetchedParse: + sessions: list[ParsedSession] + payload_bytes: int + revision_kind: RawRevisionKind + + +class RawParsePrefetchCache: + """Bounded, thread-safe store of parse results computed off the writer hold. + + polylogue-m6tp phase (a): the daemon's parse-stage warmer + (``polylogue.daemon.parse_prefetch.DaemonParseStage``) populates this + cache from a bounded ``ThreadPoolExecutor`` BEFORE the raw-materialization + conveyor's writer-hold pass runs. ``_parse_retained_raws`` below consults + it first and only falls back to its normal (writer-hold-resident) parse + on a miss. + + A miss is always safe: it reproduces the exact unmodified parse path, so + an empty, absent, or partially-warmed cache degrades to identical + behavior -- never incorrect behavior. This is what makes the cache purely + additive and lets every existing caller default to ``prefetch_cache=None`` + with zero change in outcome. + + Admission is capped by ``max_inflight_bytes`` (an explicit whale-memory + budget): a payload that would exceed the remaining budget is silently NOT + cached and is parsed normally, in the writer hold, when its turn comes. + """ + + def __init__(self, *, max_inflight_bytes: int) -> None: + if max_inflight_bytes < 1: + raise ValueError("max_inflight_bytes must be positive") + self._max_inflight_bytes = max_inflight_bytes + self._lock = threading.Lock() + self._entries: dict[str, _PrefetchedParse] = {} + self._inflight_bytes = 0 + + def __len__(self) -> int: + with self._lock: + return len(self._entries) + + def contains(self, raw_id: str) -> bool: + with self._lock: + return raw_id in self._entries + + def try_admit( + self, + raw_id: str, + sessions: list[ParsedSession], + *, + payload_bytes: int, + revision_kind: RawRevisionKind, + ) -> bool: + """Admit one already-parsed raw's output. False means the cache + already held ``raw_id`` or admitting it would exceed the budget -- + either way the caller's parse output is simply discarded, not an + error: the writer-held pass reparses that raw normally.""" + with self._lock: + if raw_id in self._entries: + return False + if self._inflight_bytes + payload_bytes > self._max_inflight_bytes: + return False + self._entries[raw_id] = _PrefetchedParse(sessions, payload_bytes, revision_kind) + self._inflight_bytes += payload_bytes + return True + + def pop(self, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind] | None: + """Remove and return one cached parse result, releasing its budget share.""" + with self._lock: + entry = self._entries.pop(raw_id, None) + if entry is None: + return None + self._inflight_bytes -= entry.payload_bytes + return entry.sessions, entry.payload_bytes, entry.revision_kind + + class RawRevisionReplayResourceBlockedError(RuntimeError): def __init__(self, raw_ids: list[str], limit_bytes: int, total_bytes: int) -> None: self.raw_ids = tuple(raw_ids) @@ -230,9 +306,15 @@ def _census_historical_revision_evidence( max_payload_bytes: int | None, ingest_workers: int = 1, commit_batch_size: int | None = None, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> _RevisionCensusState: """Persist a complete bounded parser census without mutating index.db. + ``prefetch_cache`` (polylogue-m6tp phase (a)), when supplied, is threaded + to ``_parse_retained_raws`` so a raw already parsed off the writer hold + is applied directly instead of reparsed here. ``None`` (every existing + caller) reproduces the exact unmodified parse path. + ``commit_batch_size`` (polylogue-amg1): when set to a positive integer, ``replace_raw_membership_census``/``bind_raw_revision`` writes for up to that many raws share one source.db commit instead of one commit per raw @@ -393,7 +475,9 @@ def bind_byte_proven_older_member(raw_id: str, logical_key: str) -> None: for older_raw_id in older_raw_ids } dispatch_raw_ids = [raw_id for raw_id in parseable_raw_ids if raw_id not in head_by_older] - parsed_outcomes = _parse_retained_raws(archive, dispatch_raw_ids, ingest_workers=ingest_workers) + parsed_outcomes = _parse_retained_raws( + archive, dispatch_raw_ids, ingest_workers=ingest_workers, prefetch_cache=prefetch_cache + ) for raw_id, source_index in pending_rows: if raw_id in head_by_older: continue @@ -412,7 +496,11 @@ def bind_byte_proven_older_member(raw_id: str, logical_key: str) -> None: # bundles) -- fall back to parsing every deferred member # individually, exactly as if no chain had been proven. fallback_outcomes = ( - _parse_retained_raws(archive, unresolved, ingest_workers=ingest_workers) if unresolved else {} + _parse_retained_raws( + archive, unresolved, ingest_workers=ingest_workers, prefetch_cache=prefetch_cache + ) + if unresolved + else {} ) for older_raw_id, head_raw_id in head_by_older.items(): resolved_key = head_to_key.get(head_raw_id) @@ -442,8 +530,15 @@ def census_historical_revision_evidence( max_payload_bytes: int | None = None, ingest_workers: int = 1, commit_batch_size: int | None = None, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> RevisionCensusResult: - """Complete the source-tier census stage without applying index changes.""" + """Complete the source-tier census stage without applying index changes. + + ``prefetch_cache`` (polylogue-m6tp phase (a), default ``None``) lets a + caller (the daemon conveyor) substitute already-parsed output computed + off the writer hold for any raw it warmed ahead of time. See + ``RawParsePrefetchCache`` for the equivalence guarantee. + """ with ( ArchiveStore.open_existing(archive_root, read_only=False) as archive, _ParsedSessionSpill(archive_root, max_cached_payload_bytes=max_payload_bytes) as spill, @@ -455,6 +550,7 @@ def census_historical_revision_evidence( max_payload_bytes=max_payload_bytes, ingest_workers=ingest_workers, commit_batch_size=commit_batch_size, + prefetch_cache=prefetch_cache, ) expanded, logical_keys = archive.expand_raw_membership_selection(selected_raw_ids) _record_raw_authority_parser_census(archive_root, tuple(expanded)) @@ -700,7 +796,7 @@ def _parse_retained_raw(archive: ArchiveStore, raw_id: str) -> tuple[list[Parsed return parse_retained_raw_sessions(archive, raw_id), payload_size, kind -def _census_parse_worker( +def census_parse_worker( raw_id: str, provider_token: str, blob_hash: str, @@ -720,11 +816,15 @@ def _census_parse_worker( can apply the exact same per-raw quarantine handling as the sequential path. - Dispatched onto BOTH a ``ProcessPoolExecutor`` (GIL-build fallback, see - ``_parse_unique_retained_raws``) and a ``ThreadPoolExecutor`` (real - free-threading, see ``_parse_unique_retained_raws_via_threads``) -- the - function is identical either way; only the executor and the recreated - ``ArchiveBlobPublisher``'s process/thread affinity differ. + Dispatched onto a ``ProcessPoolExecutor`` (GIL-build fallback, see + ``_parse_unique_retained_raws``), a ``ThreadPoolExecutor`` (real + free-threading, see ``_parse_unique_retained_raws_via_threads``), and the + daemon's own off-writer-hold pre-parse ``ThreadPoolExecutor`` + (``polylogue.daemon.parse_prefetch.DaemonParseStage``, polylogue-m6tp + phase (a)) -- the function is identical every time; only the executor + and the recreated ``ArchiveBlobPublisher``'s process/thread affinity + differ. Public (not module-private) precisely so the daemon's warmer can + import and dispatch it without duplicating this parse logic. """ from polylogue.storage.blob_publication import ArchiveBlobPublisher @@ -832,6 +932,7 @@ def _parse_retained_raws( raw_ids: list[str], *, ingest_workers: int, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> dict[str, tuple[list[ParsedSession], int, RawRevisionKind] | Exception]: """Parse a batch of retained raws, deduplicating byte-identical inputs. @@ -846,17 +947,33 @@ def _parse_retained_raws( because some parsers derive identity from the path (e.g. beads workspace ids), so cross-path duplicates are deliberately NOT deduplicated. Per-row ``revision_kind`` is re-attached from each row's own descriptor. + + ``prefetch_cache`` (polylogue-m6tp phase (a)) is consulted BEFORE any of + the above: a raw_id already popped from the cache is used directly and + excluded from dedup/dispatch entirely, so it costs neither a parse nor a + process/thread-pool round trip here. Every raw_id NOT found in the cache + (including all of them, when ``prefetch_cache`` is ``None`` -- the + default for every existing caller) is parsed exactly as before. """ descriptors = {raw_id: archive.raw_revision_descriptor(raw_id) for raw_id in raw_ids} + results: dict[str, tuple[list[ParsedSession], int, RawRevisionKind] | Exception] = {} + remaining_raw_ids = raw_ids + if prefetch_cache is not None and raw_ids: + remaining_raw_ids = [] + for raw_id in raw_ids: + cached = prefetch_cache.pop(raw_id) + if cached is None: + remaining_raw_ids.append(raw_id) + else: + results[raw_id] = cached grouped: dict[tuple[str, str], list[str]] = {} - for raw_id in raw_ids: + for raw_id in remaining_raw_ids: _provider, blob_hash, source_path, _kind, _size = descriptors[raw_id] grouped.setdefault((blob_hash, source_path), []).append(raw_id) representatives = [members[0] for members in grouped.values()] unique = _parse_unique_retained_raws( archive, representatives, descriptors=descriptors, ingest_workers=ingest_workers ) - results: dict[str, tuple[list[ParsedSession], int, RawRevisionKind] | Exception] = {} for members in grouped.values(): outcome = unique[members[0]] for raw_id in members: @@ -891,7 +1008,7 @@ def _parse_unique_retained_raws_via_threads( dispatches to the thread pool regardless of payload size or aggregate bytes. - Dispatches the same ``_census_parse_worker`` function the process-pool + Dispatches the same ``census_parse_worker`` function the process-pool path uses, deliberately -- NOT ``_parse_retained_raw(archive, raw_id)`` directly. ``ArchiveStore`` lazily opens ``_source_conn`` as a plain ``sqlite3.Connection`` with the default ``check_same_thread=True`` @@ -902,7 +1019,7 @@ def _parse_unique_retained_raws_via_threads( from a worker thread (as ``_parse_retained_raw`` does) raises ``sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread`` -- confirmed empirically, not - theoretical. ``_census_parse_worker`` sidesteps this entirely: it never + theoretical. ``census_parse_worker`` sidesteps this entirely: it never touches the shared ``ArchiveStore`` or its connections, only a fresh, stateless ``ArchiveBlobPublisher`` built from primitive strings (blob root + source.db path), whose blob reads are plain filesystem I/O. @@ -921,7 +1038,7 @@ def _parse_unique_retained_raws_via_threads( for raw_id in raw_ids: provider, blob_hash, source_path, _kind, _payload_size = descriptors[raw_id] future = pool.submit( - _census_parse_worker, + census_parse_worker, raw_id, provider.value, blob_hash, @@ -1017,7 +1134,7 @@ def _parse_unique_retained_raws( for raw_id in pool_raw_ids: provider, blob_hash, source_path, _kind, _payload_size = descriptors[raw_id] future = pool.submit( - _census_parse_worker, + census_parse_worker, raw_id, provider.value, blob_hash, @@ -1212,11 +1329,13 @@ def _parse_stream(provider: Provider, payload: BinaryIO, source_path: str) -> li __all__ = [ + "RawParsePrefetchCache", "RawRevisionReplayResourceBlockedError", "RevisionBackfillResult", "RevisionCensusResult", "backfill_historical_revision_evidence", "census_historical_revision_evidence", + "census_parse_worker", "record_resource_blocked_revision_census", "uncensused_historical_revision_raw_ids", "parse_retained_raw_sessions", diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 1093a1f195..c180f60b4e 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -14,7 +14,7 @@ from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path -from typing import cast +from typing import TYPE_CHECKING, cast from polylogue.archive.raw_materialization import parsed_non_session_artifact_reason from polylogue.archive.revision_authority import ( @@ -74,6 +74,17 @@ validate_raw_replay_plan, ) +if TYPE_CHECKING: + # ``revision_backfill`` imports ``ArchiveStore``, which (via + # ``polylogue.insights.archive``) imports this module -- a real, + # not merely lint-flagged, circular import at module load time. This + # type is only ever passed through here (constructed and populated by + # ``polylogue.daemon.parse_prefetch``), never constructed, so a + # ``TYPE_CHECKING``-only import plus ``from __future__ import + # annotations`` (already active above) keeps mypy strict resolved + # without ever evaluating the import at runtime. + from polylogue.sources.revision_backfill import RawParsePrefetchCache + logger = get_logger(__name__) _MAINTENANCE_TARGET_CATALOG = build_maintenance_target_catalog() _PROBE_ONLY_EXACT_MESSAGE_ROW_LIMIT = 100_000 @@ -3872,6 +3883,93 @@ def _raw_materialization_candidate_ids( ) +def raw_materialization_pending_census_raw_ids( + config: Config, + *, + limit: int, + max_payload_bytes: int, + raw_artifact_id: str | None = None, + provider: str | None = None, + source_family: str | None = None, + source_root: Path | None = None, +) -> tuple[str, ...]: + """Read-only preview of raw ids the next census pass would parse. + + polylogue-m6tp phase (a): the daemon's parse-stage warmer calls this + BEFORE taking the writer hold, to know which raws to pre-parse. Reuses + the exact same candidate + uncensused-receipt filters as + ``repair_raw_materialization``'s own census phase, and (like + ``_raw_materialization_candidate_ids``) opens only ``mode=ro`` + connections -- no write connection, and thus no writer hold, is ever + required or taken here. + + Order matches ``_raw_materialization_candidate_ids``'s row order. The + real pass additionally narrows by component grouping (only + ``census_component_limit`` components' worth of raws are actually + censused per pass), so this is a superset preview suitable for warming a + bounded pre-parse cache ahead of time, not an exact replica of one + pass's selection -- some warmed raws may go unused this pass (harmless: + they simply remain cached for a later one) and some raws a pass selects + may not have been warmed (harmless: ``_parse_retained_raws`` falls back + to its normal parse on a cache miss). + """ + if limit < 1: + raise ValueError("limit must be positive") + archive_root = _raw_materialization_archive_root(config) + candidates = _raw_materialization_candidate_ids( + config, + raw_artifact_id=raw_artifact_id, + provider=provider, + source_family=source_family, + source_root=source_root, + ) + relevant_raw_ids = list(candidates.expanded_raw_ids or tuple(candidates.raw_ids)) + if not relevant_raw_ids: + return () + from polylogue.sources.revision_backfill import uncensused_historical_revision_raw_ids + + uncensused = uncensused_historical_revision_raw_ids( + archive_root, relevant_raw_ids, max_payload_bytes=max_payload_bytes + ) + return uncensused[:limit] + + +def raw_materialization_readonly_descriptors( + archive_root: Path, raw_ids: Sequence[str] +) -> dict[str, tuple[Provider, str, str, RawRevisionKind, int]]: + """Read-only descriptor lookup for pre-parse dispatch (no writer needed). + + Mirrors ``ArchiveStore.raw_revision_descriptor`` (same columns, same + ``provider_from_origin`` projection) but over a plain ``mode=ro`` + connection: that method requires a writable blob publisher and is not + usable before the writer hold is taken, while this preview only needs + identity columns already durable in ``raw_sessions``. + """ + raw_ids = list(raw_ids) + if not raw_ids: + return {} + archive_root = Path(archive_root) + placeholders = ",".join("?" for _ in raw_ids) + result: dict[str, tuple[Provider, str, str, RawRevisionKind, int]] = {} + with closing(sqlite3.connect(f"file:{archive_root / 'source.db'}?mode=ro", uri=True)) as conn: + rows = conn.execute( + f""" + SELECT raw_id, origin, capture_mode, lower(hex(blob_hash)), source_path, revision_kind, blob_size + FROM raw_sessions WHERE raw_id IN ({placeholders}) + """, + raw_ids, + ).fetchall() + for row in rows: + result[str(row[0])] = ( + provider_from_origin(Origin.from_string(str(row[1])), family_hint=row[2]), + str(row[3]), + str(row[4]), + RawRevisionKind(str(row[5])), + int(row[6]), + ) + return result + + def _raw_materialization_stream_safe(candidates: RawMaterializationCandidates, raw_id: str) -> bool: from polylogue.sources.dispatch import is_stream_record_provider @@ -5607,6 +5705,7 @@ def repair_raw_materialization( ingest_workers: int | None = None, commit_batch_size: int | None = None, progress_callback: ProgressCallback | None = None, + prefetch_cache: RawParsePrefetchCache | None = None, ) -> RepairResult: """Converge retained raws through typed per-session revision authority. @@ -5631,6 +5730,14 @@ def repair_raw_materialization( wider ``selected_raw_ids=None`` scope (e.g. the CLI ``ops maintenance rebuild-index`` full-archive path), not by this per-component daemon loop. + + ``prefetch_cache`` (polylogue-m6tp phase (a), default ``None``) is + threaded only to the CENSUS phase's ``census_historical_revision_evidence`` + call below -- the parse-dominant stage per the 2026-07-19 perf findings. + It is NOT threaded to the REPLAY phase's + ``backfill_historical_revision_evidence`` call further down: that call's + own parse cost (``_ParsedSessionSpill.for_raw`` over already-typed + cohorts) is a different, smaller-scope code path left for a later phase. """ if max_payload_bytes < 1: raise ValueError("max_payload_bytes must be positive") @@ -5714,6 +5821,7 @@ def repair_raw_materialization( max_payload_bytes=max_payload_bytes, ingest_workers=ingest_workers, commit_batch_size=commit_batch_size, + prefetch_cache=prefetch_cache, ) except RawRevisionReplayResourceBlockedError as exc: logger.warning( diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py index e54fed58d9..31a8672a16 100644 --- a/tests/unit/daemon/test_daemon_cli.py +++ b/tests/unit/daemon/test_daemon_cli.py @@ -7,6 +7,7 @@ import os import sqlite3 import threading +import time from pathlib import Path from types import SimpleNamespace from typing import Any, cast @@ -550,6 +551,7 @@ def fake_repair_raw_materialization( dry_run: bool, raw_artifact_limit: int, max_payload_bytes: int, + prefetch_cache: object = None, ) -> FakeResult: order.append("materialize") calls["archive_root"] = config.archive_root @@ -557,6 +559,7 @@ def fake_repair_raw_materialization( calls["dry_run"] = dry_run calls["raw_artifact_limit"] = raw_artifact_limit calls["max_payload_bytes"] = max_payload_bytes + calls["prefetch_cache"] = prefetch_cache return FakeResult() def fake_recover(config: Config) -> tuple[str, ...]: @@ -601,6 +604,7 @@ def fake_converge(config: Config, *, limit: int) -> int: "dry_run": False, "raw_artifact_limit": 11, "max_payload_bytes": daemon_cli._RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES, + "prefetch_cache": None, "recover_archive_root": tmp_path / "archive", "frontier_archive_root": tmp_path / "archive", "frontier_limit": 8, @@ -1256,6 +1260,186 @@ async def fail_run_sync(*_args: object, **_kwargs: object) -> object: asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) +def test_periodic_raw_materialization_flag_off_never_warms_parse_stage( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """polylogue-m6tp phase (a): flag off (the default) must never construct + or call the parse-stage warmer, and the ``prefetch_cache`` kwarg threaded + into ``_drain_raw_materialization_once`` must be ``None`` -- reproducing + the exact unmodified in-hold parse path.""" + from polylogue.daemon import cli as daemon_cli + + class FakeResolved: + daemon_parse_stage_split = False + + seen_prefetch_cache: list[object] = [] + + def fail_daemon_parse_stage() -> object: + pytest.fail("parse-stage warmer must not be constructed when the flag is off") + + async def fake_run_sync(_actor: str, func: object, *_args: object, **_kwargs: object) -> object: + partial = cast(functools.partial[object], func) + seen_prefetch_cache.append(partial.keywords["prefetch_cache"]) + raise asyncio.CancelledError + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_parse_stage", fail_daemon_parse_stage) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fake_run_sync)) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert seen_prefetch_cache == [None] + + +def test_periodic_raw_materialization_flag_on_warms_off_writer_lease_before_drain( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """polylogue-m6tp phase (a) anti-vacuity: the parse-stage warmer must run + BEFORE the writer hold is acquired, and the drain pass must run WHILE the + lease IS held -- proven against the REAL ``DaemonWriteCoordinator`` / + ``daemon_write_lease_active`` machinery (a mock coordinator would + trivially report ``False`` for both, proving nothing). Reverting the + warm-before-run_sync ordering, or moving the warm call inside + ``run_sync``, would flip one of these two assertions.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.daemon.write_coordinator import DaemonWriteCoordinator, daemon_write_lease_active + from polylogue.product.raw_authority import RawMaterializationCounts + + class FakeResolved: + daemon_parse_stage_split = True + + order: list[str] = [] + lease_during_warm: list[bool] = [] + lease_during_drain: list[bool] = [] + _sentinel_cache = object() + + class FakeStage: + cache = _sentinel_cache + + def warm(self, config: object, *, limit: int, max_payload_bytes: int) -> int: + order.append("warm") + lease_during_warm.append(daemon_write_lease_active()) + return 0 + + def fake_drain(*, limit: int, recover: bool, prefetch_cache: object = None) -> RawMaterializationCounts: + order.append("drain") + lease_during_drain.append(daemon_write_lease_active()) + assert prefetch_cache is _sentinel_cache + raise asyncio.CancelledError + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr("polylogue.paths.render_root", lambda: tmp_path / "render") + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_parse_stage", lambda: FakeStage()) + monkeypatch.setattr(daemon_cli, "_drain_raw_materialization_once", fake_drain) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: DaemonWriteCoordinator()) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert order == ["warm", "drain"] + assert lease_during_warm == [False] + assert lease_during_drain == [True] + + +def test_periodic_raw_materialization_flag_on_warm_exception_still_hands_back_cache( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """CodeRabbit test-gap (PR #3168): ``_maybe_warm_raw_materialization_parse_stage`` + catches any exception ``stage.warm`` raises and falls back to returning + ``stage.cache`` -- whatever partial progress the warmer made before + failing -- rather than crashing the pass or discarding the cache back to + ``None``. A warm failure must degrade to "parse fewer raws off the + writer hold than usual", never "the drain pass never runs" or "every + cache-warmed raw this tick is silently lost".""" + from polylogue.daemon import cli as daemon_cli + + class FakeResolved: + daemon_parse_stage_split = True + + _sentinel_cache = object() + seen_prefetch_cache: list[object] = [] + + class FakeStage: + cache = _sentinel_cache + + def warm(self, config: object, *, limit: int, max_payload_bytes: int) -> int: + raise RuntimeError("simulated parse-stage prefetch failure") + + async def fake_run_sync(_actor: str, func: object, *_args: object, **_kwargs: object) -> object: + partial = cast(functools.partial[object], func) + seen_prefetch_cache.append(partial.keywords["prefetch_cache"]) + raise asyncio.CancelledError + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr("polylogue.paths.render_root", lambda: tmp_path / "render") + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_parse_stage", lambda: FakeStage()) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fake_run_sync)) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + # The exception from warm() never propagated (it would have surfaced as + # something other than CancelledError above), and the drain pass still + # received the stage's cache object -- not None, not a fresh cache. + assert seen_prefetch_cache == [_sentinel_cache] + + +def test_periodic_raw_materialization_flag_on_writer_hold_excludes_parse_stage_warm_time( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """The writer hold must stay short even when the parse-stage warm step is + slow: ``hold_seconds`` (measured by the REAL ``DaemonWriteCoordinator`` + around the drain call only) must be far smaller than the warm delay, + proving the write window genuinely excludes parse time rather than just + reordering it.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.daemon.write_coordinator import DaemonWriteCoordinator, DaemonWriteEvent + from polylogue.product.raw_authority import RawMaterializationCounts + + class FakeResolved: + daemon_parse_stage_split = True + + warm_delay_seconds = 0.2 + released_hold_seconds: list[float] = [] + + class FakeStage: + cache = None + + def warm(self, config: object, *, limit: int, max_payload_bytes: int) -> int: + time.sleep(warm_delay_seconds) + return 0 + + def fake_drain(*, limit: int, recover: bool, prefetch_cache: object = None) -> RawMaterializationCounts: + raise asyncio.CancelledError + + def observe(event: DaemonWriteEvent) -> None: + if event.phase == "released" and event.hold_seconds is not None: + released_hold_seconds.append(event.hold_seconds) + + monkeypatch.setattr("polylogue.config.load_polylogue_config", lambda: FakeResolved()) + monkeypatch.setattr("polylogue.paths.archive_root", lambda: tmp_path) + monkeypatch.setattr("polylogue.paths.render_root", lambda: tmp_path / "render") + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_daemon_parse_stage", lambda: FakeStage()) + monkeypatch.setattr(daemon_cli, "_drain_raw_materialization_once", fake_drain) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: DaemonWriteCoordinator(observer=observe)) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert len(released_hold_seconds) == 1 + assert released_hold_seconds[0] < warm_delay_seconds / 2 + + def test_spool_pending_check_ignores_terminal_cursor_states( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, diff --git a/tests/unit/daemon/test_parse_prefetch.py b/tests/unit/daemon/test_parse_prefetch.py new file mode 100644 index 0000000000..3bec8235b5 --- /dev/null +++ b/tests/unit/daemon/test_parse_prefetch.py @@ -0,0 +1,224 @@ +"""Tests for the daemon-owned parse-stage warmer (polylogue-m6tp phase (a)). + +Production dependencies exercised here: + +* ``DaemonParseStage.warm`` -- the actual off-writer-hold pre-parse entry + point the daemon conveyor calls. +* ``polylogue.storage.repair.raw_materialization_pending_census_raw_ids`` / + ``raw_materialization_readonly_descriptors`` -- the read-only candidate + and descriptor lookups ``warm`` uses. +* ``polylogue.sources.revision_backfill.census_parse_worker`` -- the same + pure parse function the production census path dispatches; these tests + prove it is genuinely reached via a background thread, not called inline. +""" + +from __future__ import annotations + +import sqlite3 +import threading +import time +from pathlib import Path + +import pytest + +from polylogue.config import Config +from polylogue.core.enums import Provider +from polylogue.daemon.parse_prefetch import DaemonParseStage +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + +def _codex_payload(native_id: str, text: str) -> bytes: + return ( + b'{"type":"session_meta","payload":{"id":"' + native_id.encode() + b'"}}\n' + b'{"type":"response_item","payload":{"type":"message","id":"' + + native_id.encode() + + b'-m0","role":"user","content":[{"type":"input_text","text":"' + + text.encode() + + b'"}]}}\n' + ) + + +def _config(root: Path) -> Config: + return Config(archive_root=root, render_root=root / "render", sources=[]) + + +def _seed_raws(tmp_path: Path, payloads: dict[str, bytes]) -> None: + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + for index, (source_path, payload) in enumerate(payloads.items()): + archive.write_raw_payload( + provider=Provider.CODEX, + payload=payload, + source_path=source_path, + acquired_at_ms=index, + ) + + +def test_warm_parses_pending_candidates_off_writer_hold(tmp_path: Path) -> None: + """Two never-censused raws are both discovered read-only and parsed.""" + _seed_raws( + tmp_path, + { + "a.jsonl": _codex_payload("session-a", "hello from a"), + "b.jsonl": _codex_payload("session-b", "hello from b"), + }, + ) + + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=10_000_000) + try: + warmed = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + finally: + stage.shutdown() + + assert warmed == 2 + assert len(stage.cache) == 2 + + with sqlite3.connect(f"file:{tmp_path / 'source.db'}?mode=ro", uri=True) as conn: + rows = list(conn.execute("SELECT raw_id, source_path FROM raw_sessions ORDER BY raw_id")) + raw_ids_by_path = {str(row[1]): str(row[0]) for row in rows} + + for source_path in ("a.jsonl", "b.jsonl"): + raw_id = raw_ids_by_path[source_path] + assert stage.cache.contains(raw_id) + sessions, payload_bytes, _kind = stage.cache.pop(raw_id) # type: ignore[misc] + assert len(sessions) == 1 + assert payload_bytes > 0 + + +def test_warm_dispatches_to_worker_threads_not_the_caller_thread( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Anti-vacuity: the parse must genuinely run on the pool's own thread. + + If the extraction regressed to calling the parser inline on the calling + thread (defeating the entire point of moving parse off the writer + hold), the recorded thread name would equal the caller's thread name + and this assertion would fail. + """ + _seed_raws(tmp_path, {"a.jsonl": _codex_payload("session-a", "hello")}) + + from polylogue.sources import revision_backfill + + observed_thread_names: list[str] = [] + real_worker = revision_backfill.census_parse_worker + + def spying_worker(*args: object, **kwargs: object) -> object: + observed_thread_names.append(threading.current_thread().name) + return real_worker(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(revision_backfill, "census_parse_worker", spying_worker) + + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=10_000_000) + try: + warmed = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + finally: + stage.shutdown() + + assert warmed == 1 + assert len(observed_thread_names) == 1 + assert observed_thread_names[0] != threading.current_thread().name + assert observed_thread_names[0].startswith("polylogue-parse-stage") + + +def test_warm_enforces_inflight_bytes_budget(tmp_path: Path) -> None: + """A tiny budget admits only what fits; the rest is simply left uncached + (the writer-held pass will parse it normally -- never an error).""" + _seed_raws( + tmp_path, + { + "a.jsonl": _codex_payload("session-a", "x" * 500), + "b.jsonl": _codex_payload("session-b", "y" * 500), + }, + ) + + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=1) + try: + warmed = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + finally: + stage.shutdown() + + # Neither payload fits a 1-byte budget. + assert warmed == 0 + assert len(stage.cache) == 0 + + +def test_warm_skips_raws_already_present_in_cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A second warm pass over the same still-uncensused backlog must not + redispatch a raw the cache already holds.""" + _seed_raws(tmp_path, {"a.jsonl": _codex_payload("session-a", "hello")}) + + from polylogue.sources import revision_backfill + + dispatch_count = 0 + real_worker = revision_backfill.census_parse_worker + + def counting_worker(*args: object, **kwargs: object) -> object: + nonlocal dispatch_count + dispatch_count += 1 + return real_worker(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(revision_backfill, "census_parse_worker", counting_worker) + + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=10_000_000) + try: + first = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + second = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + finally: + stage.shutdown() + + assert first == 1 + assert second == 0 + assert dispatch_count == 1 + + +def test_warm_returns_zero_when_no_candidates_pending(tmp_path: Path) -> None: + initialize_active_archive_root(tmp_path) + + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=10_000_000) + try: + warmed = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + finally: + stage.shutdown() + + assert warmed == 0 + assert len(stage.cache) == 0 + + +def test_warm_times_out_on_a_hung_worker_and_leaves_it_uncached( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """CodeRabbit (PR #3168): a worker that never returns (e.g. an + unresponsive filesystem read) must not block ``warm()`` forever -- since + ``warm()`` is awaited directly ahead of ``run_sync`` in the periodic + conveyor loop, an unbounded wait here would stall every subsequent drain + pass indefinitely. The hung raw is simply left uncached; a real + writer-held pass would reparse it normally, identical to any other + prefetch miss.""" + _seed_raws(tmp_path, {"a.jsonl": _codex_payload("session-a", "hello")}) + + from polylogue.sources import revision_backfill + + dispatched = threading.Event() + + def hanging_worker(*args: object, **kwargs: object) -> object: + dispatched.set() + time.sleep(0.3) # far longer than the test's tiny warm timeout below + pytest.fail("hung worker must not be awaited past the warm() timeout") + + monkeypatch.setattr(revision_backfill, "census_parse_worker", hanging_worker) + + stage = DaemonParseStage(max_workers=1, max_inflight_bytes=10_000_000, warm_timeout_seconds=0.02) + try: + started = time.monotonic() + warmed = stage.warm(_config(tmp_path), limit=10, max_payload_bytes=10_000_000) + elapsed = time.monotonic() - started + finally: + stage.shutdown() + + assert dispatched.wait(timeout=1.0) + assert warmed == 0 + assert len(stage.cache) == 0 + # warm() returned close to its own timeout, not after the hung worker's + # sleep -- proving the wait is genuinely bounded, not merely reordered. + assert elapsed < 0.3 diff --git a/tests/unit/daemon/test_raw_materialization_parse_stage_equivalence.py b/tests/unit/daemon/test_raw_materialization_parse_stage_equivalence.py new file mode 100644 index 0000000000..f184fd8b4a --- /dev/null +++ b/tests/unit/daemon/test_raw_materialization_parse_stage_equivalence.py @@ -0,0 +1,154 @@ +"""Equivalence: parse-stage prefetch (flag on) vs. in-hold parse (flag off). + +polylogue-m6tp phase (a). ``RawParsePrefetchCache`` is purely additive by +construction (see its docstring and the unit-level cache-hit/miss tests in +``tests/unit/sources/test_revision_backfill.py``); this test proves the +end-to-end claim against a real archive: running the SAME raw-materialization +convergence over the SAME fixture corpus, once with the daemon's +``DaemonParseStage`` warming the census parse ahead of time and once with no +prefetch cache at all, produces byte-identical durable archive content. + +Production dependencies exercised: ``DaemonParseStage.warm`` (the actual +off-writer-hold pre-parse path) feeding ``polylogue.storage.repair. +repair_raw_materialization``'s ``prefetch_cache`` parameter (the actual +production plumbing the daemon conveyor uses), not a reimplementation of +either. +""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any + +from polylogue.config import Config +from polylogue.core.enums import Provider +from polylogue.daemon.parse_prefetch import DaemonParseStage +from polylogue.storage.repair import repair_raw_materialization +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + +_VOLATILE_COLUMNS: dict[str, frozenset[str]] = { + "raw_revision_heads": frozenset({"decided_at_ms"}), + "raw_sessions": frozenset({"parsed_at_ms"}), +} + + +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-19T00: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) -> None: + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + for index in range(4): + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session( + f"session-{index}", + (("user", f"question {index}"), ("assistant", f"answer {index}")), + ), + source_path=f"corpus-{index}.jsonl", + acquired_at_ms=index, + ) + + +def _connect(path: Path) -> sqlite3.Connection: + conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + return conn + + +def _table_rows(conn: sqlite3.Connection, table: str) -> tuple[tuple[Any, ...], ...]: + excluded = _VOLATILE_COLUMNS.get(table, frozenset()) + columns = tuple( + row["name"] for row in conn.execute(f'PRAGMA table_xinfo("{table}")') if row["name"] not in excluded + ) + quoted = ", ".join(f'"{column}"' for column in columns) + rows = tuple( + sorted( + ( + tuple(bytes(value).hex() if isinstance(value, bytes) else value for value in row) + for row in conn.execute(f'SELECT {quoted} FROM "{table}"') + ), + key=repr, + ) + ) + return rows + + +def _canonical_snapshot(root: Path) -> dict[str, tuple[tuple[Any, ...], ...]]: + snapshot: dict[str, tuple[tuple[Any, ...], ...]] = {} + with _connect(root / "index.db") as conn: + for table in ("sessions", "messages", "blocks", "raw_revision_heads"): + snapshot[f"index.{table}"] = _table_rows(conn, table) + with _connect(root / "source.db") as conn: + for table in ("raw_sessions", "raw_authority_parser_census"): + snapshot[f"source.{table}"] = _table_rows(conn, table) + return snapshot + + +def test_flag_on_prefetch_and_flag_off_produce_identical_archive_content(tmp_path: Path) -> None: + baseline_root = tmp_path / "baseline" + prefetch_root = tmp_path / "prefetch" + _seed_corpus(baseline_root) + _seed_corpus(prefetch_root) + + # Flag OFF: parse happens entirely inside repair_raw_materialization, + # exactly as production behaves today. + baseline_result = repair_raw_materialization( + _config(baseline_root), + dry_run=False, + raw_artifact_limit=100, + max_payload_bytes=10_000_000, + ) + assert baseline_result.success is True + + # Flag ON: warm the SAME candidates off any writer hold first, exactly as + # the daemon's ``_maybe_warm_raw_materialization_parse_stage`` does, then + # thread the warmed cache into the identical production entry point. + stage = DaemonParseStage(max_workers=2, max_inflight_bytes=10_000_000) + try: + warmed = stage.warm(_config(prefetch_root), limit=100, max_payload_bytes=10_000_000) + assert warmed == 4 + prefetch_result = repair_raw_materialization( + _config(prefetch_root), + dry_run=False, + raw_artifact_limit=100, + max_payload_bytes=10_000_000, + prefetch_cache=stage.cache, + ) + finally: + stage.shutdown() + assert prefetch_result.success is True + # Every warmed entry was consumed by the census phase, not left stranded. + assert len(stage.cache) == 0 + + assert _canonical_snapshot(baseline_root) == _canonical_snapshot(prefetch_root) + with _connect(baseline_root / "index.db") as conn: + assert int(conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]) == 4 diff --git a/tests/unit/sources/test_revision_backfill.py b/tests/unit/sources/test_revision_backfill.py index 902416dfef..54493e1056 100644 --- a/tests/unit/sources/test_revision_backfill.py +++ b/tests/unit/sources/test_revision_backfill.py @@ -15,6 +15,7 @@ from polylogue.sources.dispatch import parse_payload from polylogue.sources.parsers.base import ParsedSession from polylogue.sources.revision_backfill import ( + RawParsePrefetchCache, _parse_one, backfill_historical_revision_evidence, census_historical_revision_evidence, @@ -866,7 +867,7 @@ def _state_db_bytes_for_session(tmp_path: Path, *, session_id: str, message_text def test_parallel_census_threads_hermes_sqlite_payload_path(tmp_path: Path) -> None: """Regression for the #3113/polylogue-1zex SQLite-detection branch under - parallel dispatch: _census_parse_worker must thread payload_path (the + parallel dispatch: census_parse_worker must thread payload_path (the real on-disk blob path) and archive_root through to _parse_one the same way the sequential parse_retained_raw_sessions does, so a Hermes state.db raw parsed by a pool worker still opens via sqlite3 against a @@ -1166,6 +1167,141 @@ def failing_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], in assert results["dup-2"] is results["dup-1"] +def test_raw_parse_prefetch_cache_admits_pops_and_enforces_budget() -> None: + """polylogue-m6tp phase (a): the daemon's warmed-parse cache is bounded + by an explicit inflight-bytes budget (the design's whale-memory guard) + and never double-admits or double-serves the same raw_id.""" + cache = RawParsePrefetchCache(max_inflight_bytes=100) + assert len(cache) == 0 + assert cache.contains("r1") is False + + assert cache.try_admit("r1", [], payload_bytes=60, revision_kind=RawRevisionKind.FULL) is True + assert len(cache) == 1 + assert cache.contains("r1") is True + + # Re-admitting the same raw_id is rejected even though the payload is small. + assert cache.try_admit("r1", [], payload_bytes=1, revision_kind=RawRevisionKind.FULL) is False + assert len(cache) == 1 + + # Budget: 60 already committed, 100 total -- a 50-byte entry would exceed it. + assert cache.try_admit("r2", [], payload_bytes=50, revision_kind=RawRevisionKind.FULL) is False + assert cache.contains("r2") is False + + # A smaller entry that fits the remaining 40 bytes is admitted. + assert cache.try_admit("r2", [], payload_bytes=40, revision_kind=RawRevisionKind.UNKNOWN) is True + assert len(cache) == 2 + + popped = cache.pop("r1") + assert popped is not None + sessions, payload_bytes, revision_kind = popped + assert sessions == [] + assert payload_bytes == 60 + assert revision_kind == RawRevisionKind.FULL + assert len(cache) == 1 + assert cache.contains("r1") is False + + # Popping releases the budget: a raw that didn't fit before now fits. + assert cache.try_admit("r3", [], payload_bytes=59, revision_kind=RawRevisionKind.FULL) is True + + # Popping an absent raw_id is a no-op, not an error. + assert cache.pop("does-not-exist") is None + + +def test_raw_parse_prefetch_cache_rejects_non_positive_budget() -> None: + with pytest.raises(ValueError, match="positive"): + RawParsePrefetchCache(max_inflight_bytes=0) + + +def test_parse_retained_raws_prefetch_cache_hit_skips_parse_entirely(monkeypatch: pytest.MonkeyPatch) -> None: + """A raw_id already popped from the prefetch cache must reach the caller's + result dict WITHOUT ever calling the parser -- proving the parse-stage + extraction actually removes that raw from the writer-hold parse path, + not merely duplicates the work. Reverting the cache-check in + ``_parse_retained_raws`` would make ``parsed`` include ``"warm-1"`` and + fail this test.""" + descriptors = { + "warm-1": (Provider.CODEX, "hash-A", "warm.jsonl", RawRevisionKind.FULL, 10), + "cold-1": (Provider.CODEX, "hash-B", "cold.jsonl", RawRevisionKind.FULL, 20), + } + + class FakeArchive: + def raw_revision_descriptor(self, raw_id: str) -> tuple[Provider, str, str, RawRevisionKind, int]: + return descriptors[raw_id] + + parsed: list[str] = [] + + def fake_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind]: + parsed.append(raw_id) + descriptor = descriptors[raw_id] + return [], descriptor[4], descriptor[3] + + monkeypatch.setattr(revision_backfill, "_parse_retained_raw", fake_parse) + + warmed_session = ParsedSession( + provider_session_id="warmed", + source_name=Provider.CODEX, + title=None, + created_at=None, + updated_at=None, + messages=[], + ) + cache = RawParsePrefetchCache(max_inflight_bytes=1_000_000) + assert cache.try_admit("warm-1", [warmed_session], payload_bytes=10, revision_kind=RawRevisionKind.FULL) is True + + results = revision_backfill._parse_retained_raws( + FakeArchive(), # type: ignore[arg-type] + ["warm-1", "cold-1"], + ingest_workers=1, + prefetch_cache=cache, + ) + + # Only the cold (unwarmed) raw actually went through the parser. + assert parsed == ["cold-1"] + assert results["warm-1"] == ([warmed_session], 10, RawRevisionKind.FULL) + sessions, size, kind = results["cold-1"] # type: ignore[misc] + assert (sessions, size, kind) == ([], 20, RawRevisionKind.FULL) + # The cache entry was consumed, not merely peeked. + assert cache.contains("warm-1") is False + + +def test_parse_retained_raws_prefetch_cache_miss_is_byte_identical_to_no_cache( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Equivalence guarantee: an empty/absent prefetch cache must produce the + exact same result as ``prefetch_cache=None`` for every existing caller + (polylogue-m6tp phase (a) is purely additive).""" + descriptors = { + "dup-1": (Provider.CODEX, "hash-A", "same.jsonl", RawRevisionKind.FULL, 10), + "dup-2": (Provider.CODEX, "hash-A", "same.jsonl", RawRevisionKind.UNKNOWN, 10), + "other-bytes": (Provider.CODEX, "hash-B", "same.jsonl", RawRevisionKind.FULL, 20), + } + + class FakeArchive: + def raw_revision_descriptor(self, raw_id: str) -> tuple[Provider, str, str, RawRevisionKind, int]: + return descriptors[raw_id] + + def fake_parse(archive: object, raw_id: str) -> tuple[list[ParsedSession], int, RawRevisionKind]: + descriptor = descriptors[raw_id] + return [], descriptor[4], descriptor[3] + + monkeypatch.setattr(revision_backfill, "_parse_retained_raw", fake_parse) + + baseline = revision_backfill._parse_retained_raws( + FakeArchive(), # type: ignore[arg-type] + list(descriptors), + ingest_workers=1, + prefetch_cache=None, + ) + with_empty_cache = revision_backfill._parse_retained_raws( + FakeArchive(), # type: ignore[arg-type] + list(descriptors), + ingest_workers=1, + prefetch_cache=RawParsePrefetchCache(max_inflight_bytes=1_000_000), + ) + + assert baseline == with_empty_cache + + def test_pool_dispatch_floor_rejects_small_aggregate_batches(monkeypatch: pytest.MonkeyPatch) -> None: """Worker spawn+import (~1.5-2s each, measured live 2026-07-19) dominates tiny batches: a per-cohort census batch of a few sub-256KiB raws must parse @@ -1351,7 +1487,7 @@ def fake_worker( ) -> tuple[str, list[ParsedSession] | None, str | None]: return raw_id, [], None - monkeypatch.setattr(revision_backfill, "_census_parse_worker", fake_worker) + monkeypatch.setattr(revision_backfill, "census_parse_worker", fake_worker) results = revision_backfill._parse_unique_retained_raws_via_threads( _NoMethodsArchive(), # type: ignore[arg-type] @@ -1398,7 +1534,7 @@ def fake_worker( raise RuntimeError(f"boom {raw_id}") return raw_id, [], None - monkeypatch.setattr(revision_backfill, "_census_parse_worker", fake_worker) + monkeypatch.setattr(revision_backfill, "census_parse_worker", fake_worker) results = revision_backfill._parse_unique_retained_raws_via_threads( _FakeArchive(), # type: ignore[arg-type] @@ -1449,7 +1585,7 @@ def fake_worker( time.sleep(delay_by_raw_id[raw_id]) return raw_id, [], None - monkeypatch.setattr(revision_backfill, "_census_parse_worker", fake_worker) + monkeypatch.setattr(revision_backfill, "census_parse_worker", fake_worker) results = revision_backfill._parse_unique_retained_raws_via_threads( _FakeArchive(), # type: ignore[arg-type]