diff --git a/docs/configuration.md b/docs/configuration.md index ceee48b00..2265b13b2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -351,6 +351,7 @@ A few keys not shown in the full example above, with their TOML path: | `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). | | `raw_authority_commit_batch_size` | `pipeline.raw_authority.commit_batch_size` | Census-phase commit batch size for raw-materialization repair (polylogue-amg1); unset uses the built-in default, `<=0` disables batching (per-raw commits). | +| `raw_authority_whale_payload_bytes` | `pipeline.raw_authority.whale_payload_bytes` | Escalation-tier payload envelope (bytes) for the daemon whale pass (polylogue-t93b); unset/`<=0` uses the built-in default (8 GiB). Widens the resource-block envelope for one dedicated, stream-safe-gated single-component pass only -- the ordinary fast-path envelope is unaffected. | | `revision_parse_dispatch_max_bytes` | `pipeline.revision_parse.dispatch_max_bytes` | Payload-size ceiling (bytes) above which a raw parses sequentially in-process instead of dispatching to the parse pool (polylogue-amg1). | | `revision_parse_pool_min_bytes` | `pipeline.revision_parse.pool_min_bytes` | Aggregate pool-eligible payload floor (bytes) below which pool dispatch cannot amortize worker-spawn cost (polylogue-amg1 / polylogue-crd8). | | `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. | @@ -359,6 +360,7 @@ A few keys not shown in the full example above, with their TOML path: | `daemon_parse_stage_max_cached_tree_bytes` | `daemon.raw_materialization.parse_stage_max_cached_tree_bytes` | Whole-cache budget (bytes) for estimated parsed-tree bytes held resident between `warm()` passes; unset/`<=0` uses the adaptive 1/8-physical-RAM default (clamped [256 MiB, 4 GiB]). | | `daemon_parse_stage_warm_timeout_seconds` | `daemon.raw_materialization.parse_stage_warm_timeout_seconds` | Bound (seconds) on how long a prefetch `warm()` pass waits for its dispatched workers; unset/`<=0` uses the 300s default. | | `daemon_bulk_rebuild_routing` | `daemon.raw_materialization.bulk_rebuild_routing` | Opt-in (polylogue-m6tp phase (c) / polylogue-gd6v, default off): once a raw backlog is bulk-scale, route it into a daemon-owned resumable blue-green index generation build instead of the trickle conveyor, promoting it once exact-ready. | +| `daemon_whale_raw_materialization` | `daemon.raw_materialization.whale_convergence` | Escalation tier (polylogue-t93b, default **on**): once the ordinary trickle conveyor is quiescent, run a dedicated bounded pass for one resource-blocked, stream-safe raw-authority component at a time instead of leaving it permanently offline-manual. Non-stream-safe oversized components remain blocked with a distinct typed reason. Set `false`/`0` to hold whale components fully offline-manual (the pre-polylogue-t93b behavior). | | `judgment_automation_enabled` | `judgment_automation.enabled` | Opt-in (polylogue-6qjc, default off): schedule the daemon judgment-automation sweep that calls the `judge` dispatcher on auto-judgeable assertion candidates per policy and escalates the rest to a `handoff` assertion. Exercises the same authority as the MCP `judge` dispatcher, so the sweep also requires `mcp_judge_enabled`. See [`docs/daemon.md`](daemon.md). | | `judgment_automation_interval_s` | `judgment_automation.interval_s` | Seconds between judgment-automation sweeps (default 3600; floored at 60 at runtime). | | `judgment_automation_batch_limit` | `judgment_automation.batch_limit` | Maximum candidates judged per judgment-automation sweep (default 200). | @@ -408,6 +410,8 @@ Common runtime overrides: | `POLYLOGUE_TOKEN_PATH` | Drive auth | OAuth token path. | | `POLYLOGUE_HOOK_PROVIDER` | `hook_provider` | Force hook-harness detection to `claude-code`/`codex`. | | `POLYLOGUE_RAW_AUTHORITY_COMMIT_BATCH_SIZE` | `raw_authority_commit_batch_size` | Census-phase commit batch size for raw-materialization repair. | +| `POLYLOGUE_RAW_AUTHORITY_WHALE_PAYLOAD_BYTES` | `raw_authority_whale_payload_bytes` | Escalation-tier payload envelope for the daemon whale pass. | +| `POLYLOGUE_DAEMON_WHALE_RAW_MATERIALIZATION` | `daemon_whale_raw_materialization` | Escalation tier on/off (default on): bounded whale pass for resource-blocked, stream-safe components. | | `POLYLOGUE_REVISION_PARSE_DISPATCH_MAX_BYTES` | `revision_parse_dispatch_max_bytes` | Payload-size ceiling for pool-eligible revision-parse dispatch. | | `POLYLOGUE_REVISION_PARSE_POOL_MIN_BYTES` | `revision_parse_pool_min_bytes` | Aggregate payload floor for revision-parse pool dispatch to amortize. | | `POLYLOGUE_DAEMON_PARSE_STAGE_WORKERS` | `daemon_parse_stage_workers` | Worker cap for the daemon-owned pre-parse thread pool. | diff --git a/polylogue/config.py b/polylogue/config.py index a1780938c..e392326c0 100644 --- a/polylogue/config.py +++ b/polylogue/config.py @@ -619,6 +619,21 @@ def raw_authority_commit_batch_size(self) -> int | None: return None return int(str(value)) + @property + def raw_authority_whale_payload_bytes(self) -> int | None: + """Escalation-tier payload envelope (bytes) for the daemon whale pass (polylogue-t93b). + + ``None``/absent falls back to the caller's hardcoded default. The + ordinary daemon fast-path blob limit is unaffected by this value -- + it only widens the envelope for a dedicated, single-component pass + (stream-safe members only) run when the ordinary trickle conveyor + is otherwise quiescent. See ``daemon_whale_raw_materialization``. + """ + value = self._data.get("raw_authority_whale_payload_bytes") + if value is None: + return None + return int(str(value)) + @property def revision_parse_dispatch_max_bytes(self) -> int | None: """Payload-size ceiling (bytes) for pool-eligible revision parse dispatch. @@ -749,6 +764,24 @@ def daemon_bulk_rebuild_routing(self) -> bool: """ return bool(self._data.get("daemon_bulk_rebuild_routing")) + @property + def daemon_whale_raw_materialization(self) -> bool: + """Escalation tier for whale-scale resource-blocked raw components (polylogue-t93b). + + On by default: a raw-authority component permanently resource-blocked + at the ordinary daemon fast-path envelope converges through a + dedicated, bounded pass (single component, stream-safe members only, + commit-batched replay via ``raw_authority_commit_batch_size``) once + the ordinary trickle conveyor is otherwise quiescent -- a permanent + offline-only requirement is a policy bug per the automagic-invariants + doctrine (operator ruling, live witness codex:019f49d8, 6.33GB/788 + raws with zero index presence). Set TOML + ``[daemon.raw_materialization] whale_convergence = false`` or + ``POLYLOGUE_DAEMON_WHALE_RAW_MATERIALIZATION=0`` to hold whale + components fully offline-manual instead. + """ + return _require_bool_config_value(self._data, "daemon_whale_raw_materialization") + @property def judgment_automation_enabled(self) -> bool: """Explicit opt-in for the daemon judgment-automation sweep (polylogue-6qjc). @@ -1375,6 +1408,19 @@ def effective_path(self) -> str: "(polylogue-amg1); <=0 disables batching (per-raw commits)." ), ), + ConfigInventoryEntry( + "raw_authority_whale_payload_bytes", + toml_path="pipeline.raw_authority.whale_payload_bytes", + env_var="POLYLOGUE_RAW_AUTHORITY_WHALE_PAYLOAD_BYTES", + owner_class="resource-policy", + reload_behavior="daemon-loop", + description=( + "Escalation-tier payload envelope (bytes) for the daemon whale " + "pass (polylogue-t93b); widens the resource-block envelope for " + "a dedicated single-component stream-safe-gated pass only. " + "<=0/absent falls back to the adaptive default." + ), + ), ConfigInventoryEntry( "revision_parse_dispatch_max_bytes", toml_path="pipeline.revision_parse.dispatch_max_bytes", @@ -1505,6 +1551,21 @@ def effective_path(self) -> str: "<=0 falls back to the 300s default." ), ), + ConfigInventoryEntry( + "daemon_whale_raw_materialization", + toml_path="daemon.raw_materialization.whale_convergence", + env_var="POLYLOGUE_DAEMON_WHALE_RAW_MATERIALIZATION", + owner_class="resource-policy", + reload_behavior="daemon-loop", + description=( + "Escalation tier (polylogue-t93b): on by default. Once the " + "ordinary trickle conveyor is quiescent, run a dedicated " + "bounded pass for one resource-blocked, stream-safe raw " + "authority component at a time instead of leaving it " + "permanently offline-manual. Non-stream-safe oversized " + "components remain blocked with a distinct typed reason." + ), + ), ConfigInventoryEntry( "daemon_bulk_rebuild_routing", toml_path="daemon.raw_materialization.bulk_rebuild_routing", @@ -1583,6 +1644,7 @@ def effective_path(self) -> str: "judgment_automation_interval_s", "judgment_automation_batch_limit", "raw_authority_commit_batch_size", + "raw_authority_whale_payload_bytes", "revision_parse_dispatch_max_bytes", "revision_parse_pool_min_bytes", "daemon_parse_stage_workers", @@ -1612,6 +1674,7 @@ def effective_path(self) -> str: "observability_enabled", "daemon_parse_stage_split", "daemon_bulk_rebuild_routing", + "daemon_whale_raw_materialization", "mcp_write_enabled", "mcp_judge_enabled", "mcp_maintenance_enabled", @@ -1813,6 +1876,7 @@ def _default_config_values(bootstrap: _BootstrapPaths | None = None) -> dict[str "ingest_parse_workers": default_parse_workers, "live_full_ingest_workers": 1, "raw_authority_commit_batch_size": None, + "raw_authority_whale_payload_bytes": None, "revision_parse_dispatch_max_bytes": None, "revision_parse_pool_min_bytes": None, "subscription_plans": (), @@ -1822,6 +1886,7 @@ def _default_config_values(bootstrap: _BootstrapPaths | None = None) -> dict[str "daemon_parse_stage_max_cached_tree_bytes": None, "daemon_parse_stage_warm_timeout_seconds": None, "daemon_bulk_rebuild_routing": False, + "daemon_whale_raw_materialization": True, "mcp_write_enabled": False, "mcp_judge_enabled": False, "mcp_maintenance_enabled": False, diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 6288953f4..e3e759dea 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -87,6 +87,17 @@ # (candidate discovery, component ordering, receipt) over more components. _RAW_MATERIALIZATION_CENSUS_BATCH_LIMIT = 64 _RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES = 64 * 1024 * 1024 +# polylogue-t93b: escalation-tier envelope for the whale pass. A component +# permanently resource-blocked at the ordinary fast-path limit above +# converges through a dedicated, single-component pass at this wider +# envelope instead -- still bounded (a genuinely unbounded component stays +# blocked), gated on every member being stream-record-safe (bounded parse +# memory via the existing RawParsePrefetchCache/whale-spill machinery) and +# on the ordinary trickle conveyor being otherwise quiescent (bounded writer +# contention). 8 GiB comfortably covers the live witness (codex:019f49d8, +# 6.33GB/788 raws) with headroom; override via +# ``raw_authority_whale_payload_bytes`` / POLYLOGUE_RAW_AUTHORITY_WHALE_PAYLOAD_BYTES. +_RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES = 8 * 1024 * 1024 * 1024 _RAW_MATERIALIZATION_LIVE_SPOOL_BACKOFF_SECONDS = 60 # A spool file younger than this is in the live route's normal debounce/ # batch flow, not stalled; only older cursor-less files park the conveyor. @@ -784,6 +795,12 @@ async def _periodic_raw_materialization_convergence() -> None: await asyncio.sleep(_RAW_MATERIALIZATION_LIVE_SPOOL_BACKOFF_SECONDS) continue recover = True + # polylogue-t93b: set True only at the genuine-quiescence break below + # (no progress AND no remaining ordinary-envelope candidates this + # tick) -- never on the bulk-rebuild-in-flight or spool-pending + # breaks, which are deferrals for unrelated reasons, not evidence the + # ordinary backlog is settled. + quiescent = False try: while True: # polylogue-gd6v residual: while the daemon's own bulk-rebuild @@ -861,6 +878,7 @@ async def _periodic_raw_materialization_convergence() -> None: materialized.remaining_candidates, ) if materialized.remaining_candidates <= 0 or not materialized.made_progress: + quiescent = True break if _browser_capture_spool_has_pending_files(): break @@ -872,6 +890,17 @@ async def _periodic_raw_materialization_convergence() -> None: logger.warning("raw materialization: convergence check failed", exc_info=True) except Exception: logger.warning("raw materialization: convergence check failed", exc_info=True) + # polylogue-t93b: escalate a resource-blocked, stream-safe component + # only once the ordinary trickle conveyor is genuinely quiescent for + # this tick (see ``quiescent`` above) and not yielding to pending + # browser-capture spool files -- keeps the whale pass from ever + # competing with ordinary-scale backlog or live-capture ingest for + # the writer hold. + if quiescent and not _browser_capture_spool_has_pending_files(): + try: + await _maybe_run_raw_materialization_whale_pass() + except Exception: + logger.warning("raw materialization: whale-pass scheduling failed", exc_info=True) await asyncio.sleep(_RAW_MATERIALIZATION_CONVERGENCE_INTERVAL_SECONDS) @@ -1025,6 +1054,142 @@ def _drain_raw_materialization_once( ) +def _resolve_raw_materialization_whale_blob_limit_bytes() -> int: + """Resolve the whale-pass escalation envelope (polylogue-t93b).""" + from polylogue.config import load_polylogue_config + + configured = load_polylogue_config().raw_authority_whale_payload_bytes + if configured is None or configured <= 0: + return _RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES + return configured + + +def _run_raw_materialization_whale_pass_once(*, raw_artifact_id: str, max_payload_bytes: int) -> Any: + """Run one bounded, single-component whale pass under the writer hold. + + polylogue-t93b. Mirrors ``_drain_raw_materialization_once`` but scoped to + exactly the logical authority component containing ``raw_artifact_id`` + (the same "one seed expands to the whole component via membership" + convention the CLI's ``--raw-artifact-id`` single-source repair already + uses), at the widened whale envelope instead of the ordinary fast-path + limit. ``raw_artifact_limit=1`` bounds selection to that one component -- + writer-hold length within it is bounded by + ``raw_authority_commit_batch_size`` (PR #3248), unchanged by this call. + + Deliberately skips the general (unscoped) interrupted-frontier recovery + and cross-archive frontier convergence that ``_drain_raw_materialization_once`` + performs: those are ordinary-pass hygiene over the whole backlog, not + specific to this one escalated component, and the ordinary pass already + runs them on its own cadence. + """ + from polylogue.config import Config + from polylogue.paths import archive_root, render_root + from polylogue.product import raw_authority + + archive = archive_root() + config = Config(archive_root=archive, render_root=render_root(), sources=[]) + try: + result = raw_authority.repair_materialization( + config, + dry_run=False, + raw_artifact_limit=1, + max_payload_bytes=max_payload_bytes, + raw_artifact_id=raw_artifact_id, + ) + finally: + _close_raw_materialization_fts(config.archive_root / "index.db") + _emit_raw_materialization_pass(result) + if not result.success: + logger.warning("raw materialization: whale pass for %s incomplete: %s", raw_artifact_id, result.detail) + return result + + +async def _maybe_run_raw_materialization_whale_pass() -> bool: + """Escalate one resource-blocked, stream-safe component when quiescent. + + polylogue-t93b: called only after the ordinary trickle conveyor has + drained to quiescence for this tick (see + ``_periodic_raw_materialization_convergence``), so a whale pass never + competes with ordinary-scale backlog for the writer hold. Off switch: + ``daemon_whale_raw_materialization`` (on by default -- a permanent + offline-only requirement for whale components is the policy bug this + closes). + + Returns whether a pass was genuinely attempted this call, mirroring + ``_maybe_route_daemon_bulk_rebuild``'s return-value contract so the + caller can decide burst-vs-outer-interval pacing the same way. + """ + from polylogue.config import load_polylogue_config + + if not load_polylogue_config().daemon_whale_raw_materialization: + return False + from polylogue.config import Config + from polylogue.daemon.events import emit_daemon_event + from polylogue.paths import archive_root, render_root + from polylogue.product import raw_authority + + root = archive_root() + config = Config(archive_root=root, render_root=render_root(), sources=[]) + whale_limit = _resolve_raw_materialization_whale_blob_limit_bytes() + try: + candidate = await asyncio.to_thread( + raw_authority.whale_pass_candidate, + config, + ordinary_max_payload_bytes=_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES, + whale_max_payload_bytes=whale_limit, + ) + except Exception: + logger.warning("raw materialization: whale-pass candidate lookup failed", exc_info=True) + return False + if candidate is None: + return False + logger.info( + "raw materialization: starting whale pass for component seeded by %s (envelope=%d bytes)", + candidate, + whale_limit, + ) + emit_daemon_event( + "raw_materialization_whale_pass_started", + payload={"seed_raw_id": candidate, "max_payload_bytes": whale_limit}, + ) + try: + result = await daemon_write_coordinator().run_sync( + "maintenance.raw_materialization_whale", + functools.partial( + _run_raw_materialization_whale_pass_once, + raw_artifact_id=candidate, + max_payload_bytes=whale_limit, + ), + ) + except sqlite3.OperationalError as exc: + if is_transient_sqlite_lock(exc): + logger.info("raw materialization: whale pass deferred, archive busy: %s", exc) + else: + logger.warning("raw materialization: whale pass failed", exc_info=True) + emit_daemon_event( + "raw_materialization_whale_pass_completed", + payload={"seed_raw_id": candidate, "success": False, "detail": str(exc)}, + ) + return True + except Exception as exc: + logger.warning("raw materialization: whale pass failed", exc_info=True) + emit_daemon_event( + "raw_materialization_whale_pass_completed", + payload={"seed_raw_id": candidate, "success": False, "detail": str(exc)}, + ) + return True + emit_daemon_event( + "raw_materialization_whale_pass_completed", + payload={ + "seed_raw_id": candidate, + "success": bool(result.success), + "repaired_count": int(result.repaired_count), + "detail": str(result.detail), + }, + ) + return True + + def _browser_capture_spool_has_pending_files() -> bool: """Whether live browser evidence is STALLED short of its ingest route. diff --git a/polylogue/product/raw_authority.py b/polylogue/product/raw_authority.py index 8315d0bd1..d18e5e657 100644 --- a/polylogue/product/raw_authority.py +++ b/polylogue/product/raw_authority.py @@ -76,6 +76,7 @@ def repair_materialization( raw_artifact_limit: int, max_payload_bytes: int, prefetch_cache: RawParsePrefetchCache | None = None, + raw_artifact_id: str | None = None, ) -> Any: """Run one bounded raw source->index convergence pass. @@ -83,6 +84,12 @@ def repair_materialization( caller substitute parse output already computed off the writer hold for this pass's census phase; see ``polylogue.sources.revision_backfill.RawParsePrefetchCache``. + + ``raw_artifact_id`` (polylogue-t93b, default ``None``) scopes the pass to + the single logical authority component containing that raw -- the + daemon's escalation-tier whale pass uses this to converge one + resource-blocked component at a time under a widened ``max_payload_bytes`` + envelope instead of re-scanning the whole archive-wide backlog. """ from polylogue.storage.repair import repair_raw_materialization @@ -92,6 +99,28 @@ def repair_materialization( raw_artifact_limit=raw_artifact_limit, max_payload_bytes=max_payload_bytes, prefetch_cache=prefetch_cache, + raw_artifact_id=raw_artifact_id, + ) + + +def whale_pass_candidate( + config: Config, + *, + ordinary_max_payload_bytes: int, + whale_max_payload_bytes: int, +) -> str | None: + """Read-only: pick one resource-blocked, stream-safe component to escalate. + + polylogue-t93b. See + ``polylogue.storage.repair.raw_materialization_whale_pass_candidate`` for + the selection contract; safe to call without the writer hold. + """ + from polylogue.storage.repair import raw_materialization_whale_pass_candidate + + return raw_materialization_whale_pass_candidate( + config, + ordinary_max_payload_bytes=ordinary_max_payload_bytes, + whale_max_payload_bytes=whale_max_payload_bytes, ) diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 3859c4222..361b25bb0 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -228,6 +228,7 @@ def record_resource_blocked_revision_census( *, max_payload_bytes: int, total_payload_bytes: int, + stream_safe: bool | None = None, ) -> None: """Persist a non-terminal no-retry receipt for immutable oversized bytes. @@ -235,13 +236,30 @@ def record_resource_blocked_revision_census( the payload. The fingerprint binds that fact to the exact admission envelope, so increasing the envelope (or changing the parser identity) re-admits the raw without a timer-driven retry storm. + + ``stream_safe`` (polylogue-t93b, default ``None`` for callers that have + not classified it) records whether every member of the blocked + component is stream-record-safe -- the daemon's escalation-tier whale + pass (``daemon_whale_raw_materialization``) only ever selects + stream-safe components, so this distinguishes "waiting for a bounded + whale pass" from "genuinely cannot converge automatically" directly in + the durable census detail, without changing the admission fingerprint + (which stays keyed on the envelope alone, preserving the existing + re-admission-on-wider-envelope invariant). """ if not raw_ids: return fingerprint = _resource_blocked_parser_fingerprint(max_payload_bytes) + escalation_note = ( + "; escalation-eligible: stream-safe, awaiting a bounded daemon whale pass" + if stream_safe + else "; escalation-blocked: non-stream-safe, requires manual/offline convergence" + if stream_safe is False + else "" + ) detail = ( "current parser census deferred before blob open: " - f"component payload {total_payload_bytes} exceeds envelope {max_payload_bytes}" + f"component payload {total_payload_bytes} exceeds envelope {max_payload_bytes}{escalation_note}" ) with sqlite3.connect(archive_root / "source.db") as conn, conn: for raw_id in raw_ids: diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 576017855..306f0cf1c 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -3977,6 +3977,20 @@ def _raw_materialization_stream_safe(candidates: RawMaterializationCandidates, r return is_stream_record_provider(candidates.raw_source_paths.get(raw_id), provider) +def _raw_materialization_component_stream_safe( + candidates: RawMaterializationCandidates, component: tuple[str, ...] +) -> bool: + """Whether every member of an authority component is stream-record-safe. + + polylogue-t93b: the daemon's escalation-tier whale pass requires this to + be true for the whole component before widening the resource-block + envelope for it -- a mixed component with even one non-stream-safe + member stays permanently blocked (typed distinctly) rather than risking + an unbounded eager parse of the non-stream-safe member. + """ + return all(_raw_materialization_stream_safe(candidates, raw_id) for raw_id in component) + + def _raw_materialization_retryable_missing_blob_error(parse_error: object) -> bool: if not isinstance(parse_error, str): return False @@ -4033,6 +4047,55 @@ def _raw_materialization_component_blob_bytes(candidates: RawMaterializationCand return candidates.expanded_blob_bytes.get(raw_id, candidates.raw_blob_bytes.get(raw_id, 0)) +def raw_materialization_whale_pass_candidate( + config: Config, + *, + ordinary_max_payload_bytes: int, + whale_max_payload_bytes: int, +) -> str | None: + """Read-only: pick one component for the daemon's escalation-tier whale pass. + + polylogue-t93b: the ordinary fast-path envelope (``ordinary_max_payload_bytes``, + the daemon's ``_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES``) permanently + resource-blocks any component whose aggregate payload exceeds it -- a + policy bug per the automagic-invariants doctrine when the daemon owns + raw->index convergence. This selects the highest-fairness-priority + component that is: + + - oversized under the ordinary envelope (otherwise the regular trickle + pass already converges it, no escalation needed); + - within the wider ``whale_max_payload_bytes`` escalation envelope + (still bounded -- a genuinely unbounded component stays blocked); and + - entirely stream-record-safe (every member), so the parse dispatch that + already exists for stream-record providers keeps parse memory bounded + without any new streaming code path. + + Returns the component's seed raw id (the same "one seed expands to the + whole logical component via membership" convention + ``repair_raw_materialization``'s ``raw_artifact_id`` scoping already + uses), or ``None`` if no component currently qualifies. Opens only + ``mode=ro`` connections -- safe to call without the writer hold, exactly + like ``raw_materialization_pending_census_raw_ids``. + """ + archive_root = _raw_materialization_archive_root(config) + if not (archive_root / "source.db").exists() or not (archive_root / "index.db").exists(): + return None + candidates = _raw_materialization_candidate_ids(config) + if not candidates.raw_ids: + return None + ordered_components = _raw_materialization_ordered_components(candidates, archive_root=archive_root) + for component in ordered_components: + total_bytes = sum(_raw_materialization_component_blob_bytes(candidates, member) for member in component) + if total_bytes <= ordinary_max_payload_bytes: + continue + if total_bytes > whale_max_payload_bytes: + continue + if not _raw_materialization_component_stream_safe(candidates, component): + continue + return _raw_materialization_component_seed(candidates, component) + return None + + def _raw_authority_scope( *, raw_artifact_id: str | None, @@ -5844,6 +5907,7 @@ def repair_raw_materialization( tuple(sorted(uncensused_raw_ids.intersection(component))), max_payload_bytes=max_payload_bytes, total_payload_bytes=exc.total_bytes, + stream_safe=_raw_materialization_component_stream_safe(candidates, component), ) except Exception: logger.exception("raw authority census failed for component containing %s", seed) @@ -5918,6 +5982,14 @@ def repair_raw_materialization( } if stream_oversized: metrics["raw_materialization_stream_oversized_count"] = float(len(stream_oversized)) + non_stream_oversized = oversized - stream_oversized + if non_stream_oversized: + # polylogue-t93b: distinguishes "waiting for a bounded + # daemon whale pass" (stream-safe) from "genuinely + # cannot converge automatically" (non-stream-safe) at + # the census surface, mirroring the durable detail text + # persisted by ``record_resource_blocked_revision_census``. + metrics["raw_materialization_non_stream_safe_oversized_count"] = float(len(non_stream_oversized)) detail = ( f"Raw replay planning paused until the persisted parser census completes for " f"{len(census_pending_raw_ids):,} relevant raw(s)" @@ -6023,6 +6095,14 @@ def repair_raw_materialization( metrics["raw_materialization_execute_blob_limit_bytes"] = float(max_payload_bytes) if oversized_stream_safe_raw_ids: metrics["raw_materialization_stream_oversized_count"] = float(len(oversized_stream_safe_raw_ids)) + oversized_non_stream_safe_raw_ids = [ + raw_id for raw_id in oversized_candidate_raw_ids if raw_id not in set(oversized_stream_safe_raw_ids) + ] + if oversized_non_stream_safe_raw_ids: + # polylogue-t93b: complements the stream-safe count so the census + # distinguishes escalation-eligible from genuinely-blocked oversized + # candidates at this (post-plan) surface too. + metrics["raw_materialization_non_stream_safe_oversized_count"] = float(len(oversized_non_stream_safe_raw_ids)) if deferred_plan_ids: metrics["raw_materialization_deferred_plan_count"] = float(len(deferred_plan_ids)) scope = _raw_authority_scope( diff --git a/tests/unit/daemon/test_daemon_cli.py b/tests/unit/daemon/test_daemon_cli.py index 3cddae834..d9a57f68b 100644 --- a/tests/unit/daemon/test_daemon_cli.py +++ b/tests/unit/daemon/test_daemon_cli.py @@ -552,6 +552,7 @@ def fake_repair_raw_materialization( raw_artifact_limit: int, max_payload_bytes: int, prefetch_cache: object = None, + raw_artifact_id: str | None = None, ) -> FakeResult: order.append("materialize") calls["archive_root"] = config.archive_root @@ -560,6 +561,7 @@ def fake_repair_raw_materialization( calls["raw_artifact_limit"] = raw_artifact_limit calls["max_payload_bytes"] = max_payload_bytes calls["prefetch_cache"] = prefetch_cache + calls["raw_artifact_id"] = raw_artifact_id return FakeResult() def fake_recover(config: Config) -> tuple[str, ...]: @@ -605,6 +607,7 @@ def fake_converge(config: Config, *, limit: int) -> int: "raw_artifact_limit": 11, "max_payload_bytes": daemon_cli._RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES, "prefetch_cache": None, + "raw_artifact_id": None, "recover_archive_root": tmp_path / "archive", "frontier_archive_root": tmp_path / "archive", "frontier_limit": 8, @@ -4200,3 +4203,196 @@ async def fake_sleep(seconds: float) -> None: assert trickle_calls == ["maintenance.raw_materialization"] assert len(routed) >= 1 # the suppression branch drove at least one bulk pass first + + +# polylogue-t93b: the daemon's whale-pass escalation tier. A component +# permanently resource-blocked at the ordinary fast-path envelope must not +# stay blocked forever -- the periodic conveyor schedules a dedicated, +# bounded pass for it once (and only once) the ordinary trickle backlog is +# genuinely quiescent for the tick. + + +def test_periodic_raw_materialization_convergence_schedules_whale_pass_on_quiescence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Quiescence (no remaining candidates, no progress) must trigger exactly + one whale-pass attempt per tick, after the trickle burst settles. + + Anti-vacuity: deleting the ``if quiescent and not ...: await + _maybe_run_raw_materialization_whale_pass()`` call this test patches + (i.e. reverting the daemon wiring in polylogue/daemon/cli.py) makes + ``whale_calls == [1]`` fail with an empty list -- verified by running + this exact assertion against the unmodified pre-patch source tree + during implementation. + """ + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + whale_calls: list[int] = [] + + async def fake_whale_pass() -> bool: + whale_calls.append(1) + return False + + async def fake_run_sync(_actor: str, _func: object, *_args: object, **_kwargs: object) -> object: + return RawMaterializationCounts(remaining_candidates=0) + + sleeps: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + if seconds == daemon_cli._RAW_MATERIALIZATION_CONVERGENCE_INTERVAL_SECONDS: + raise asyncio.CancelledError + + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_maybe_run_raw_materialization_whale_pass", fake_whale_pass) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fake_run_sync)) + + with patch("asyncio.sleep", side_effect=fake_sleep), pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert whale_calls == [1] + + +def test_periodic_raw_materialization_convergence_skips_whale_pass_mid_burst( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """While the ordinary trickle conveyor is still making progress (not + quiescent), the whale pass must not be attempted -- it only runs once + the ordinary backlog settles for the tick.""" + from polylogue.daemon import cli as daemon_cli + from polylogue.product.raw_authority import RawMaterializationCounts + + whale_calls: list[int] = [] + + async def fake_whale_pass() -> bool: + whale_calls.append(1) + return False + + remaining_schedule = [5, 0] + + async def fake_run_sync(_actor: str, _func: object, *_args: object, **_kwargs: object) -> object: + remaining = remaining_schedule.pop(0) + return RawMaterializationCounts(repaired_sessions=1, remaining_candidates=remaining) + + async def fake_sleep(seconds: float) -> None: + if seconds == daemon_cli._RAW_MATERIALIZATION_CONVERGENCE_INTERVAL_SECONDS: + raise asyncio.CancelledError + + monkeypatch.setattr(daemon_cli, "_browser_capture_spool_has_pending_files", lambda: False) + monkeypatch.setattr(daemon_cli, "_maybe_run_raw_materialization_whale_pass", fake_whale_pass) + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fake_run_sync)) + + with patch("asyncio.sleep", side_effect=fake_sleep), pytest.raises(asyncio.CancelledError): + asyncio.run(daemon_cli._periodic_raw_materialization_convergence()) + + assert remaining_schedule == [] + # Quiescence is only reached on the SECOND pass (remaining=0); the whale + # pass fires exactly once, after the burst settles -- not on the first, + # still-progressing pass. + assert whale_calls == [1] + + +def test_maybe_run_raw_materialization_whale_pass_runs_scoped_pass_and_emits_events( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A found whale candidate must run through the writer coordinator + scoped to that one raw id at the resolved whale envelope, and emit + start/completion daemon events carrying the seed id and envelope.""" + from polylogue.daemon import cli as daemon_cli + + events: list[tuple[str, dict[str, object]]] = [] + monkeypatch.setattr( + "polylogue.daemon.events.emit_daemon_event", + lambda kind, *, payload: events.append((kind, payload)), + ) + monkeypatch.setattr( + "polylogue.config.load_polylogue_config", + lambda: SimpleNamespace(daemon_whale_raw_materialization=True, raw_authority_whale_payload_bytes=None), + ) + monkeypatch.setattr( + "polylogue.product.raw_authority.whale_pass_candidate", + lambda _config, **_kwargs: "whale-seed-raw-id", + ) + + run_sync_calls: list[tuple[str, str, int]] = [] + + async def fake_run_sync(actor: str, func: object, *_args: object, **_kwargs: object) -> object: + partial = cast("functools.partial[object]", func) + run_sync_calls.append((actor, partial.keywords["raw_artifact_id"], partial.keywords["max_payload_bytes"])) + return SimpleNamespace(success=True, repaired_count=3, detail="whale converged") + + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", lambda: SimpleNamespace(run_sync=fake_run_sync)) + + attempted = asyncio.run(daemon_cli._maybe_run_raw_materialization_whale_pass()) + + assert attempted is True + assert run_sync_calls == [ + ( + "maintenance.raw_materialization_whale", + "whale-seed-raw-id", + daemon_cli._RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES, + ) + ] + kinds = [kind for kind, _payload in events] + assert kinds == ["raw_materialization_whale_pass_started", "raw_materialization_whale_pass_completed"] + assert events[0][1] == { + "seed_raw_id": "whale-seed-raw-id", + "max_payload_bytes": daemon_cli._RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES, + } + assert events[1][1] == { + "seed_raw_id": "whale-seed-raw-id", + "success": True, + "repaired_count": 3, + "detail": "whale converged", + } + + +def test_maybe_run_raw_materialization_whale_pass_no_candidate_skips_writer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """No qualifying component this tick must never touch the writer + coordinator -- the read-only candidate lookup is the only thing that + runs.""" + from polylogue.daemon import cli as daemon_cli + + monkeypatch.setattr( + "polylogue.config.load_polylogue_config", + lambda: SimpleNamespace(daemon_whale_raw_materialization=True, raw_authority_whale_payload_bytes=None), + ) + monkeypatch.setattr( + "polylogue.product.raw_authority.whale_pass_candidate", + lambda _config, **_kwargs: None, + ) + + def fail_coordinator() -> object: + pytest.fail("no candidate found -- must not touch the writer coordinator") + + monkeypatch.setattr(daemon_cli, "daemon_write_coordinator", fail_coordinator) + + attempted = asyncio.run(daemon_cli._maybe_run_raw_materialization_whale_pass()) + + assert attempted is False + + +def test_maybe_run_raw_materialization_whale_pass_respects_off_switch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``daemon_whale_raw_materialization=False`` must hold a whale + component fully offline-manual (the pre-polylogue-t93b behavior) -- + neither the candidate lookup nor the writer coordinator run.""" + from polylogue.daemon import cli as daemon_cli + + monkeypatch.setattr( + "polylogue.config.load_polylogue_config", + lambda: SimpleNamespace(daemon_whale_raw_materialization=False, raw_authority_whale_payload_bytes=None), + ) + + def fail_candidate_lookup(_config: object, **_kwargs: object) -> object: + pytest.fail("off switch must short-circuit before the candidate lookup") + + monkeypatch.setattr("polylogue.product.raw_authority.whale_pass_candidate", fail_candidate_lookup) + + attempted = asyncio.run(daemon_cli._maybe_run_raw_materialization_whale_pass()) + + assert attempted is False diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py index 4e4176ed2..77b1bb504 100644 --- a/tests/unit/storage/test_repair.py +++ b/tests/unit/storage/test_repair.py @@ -2869,3 +2869,227 @@ def test_offline_maintenance_preview_allowed_with_live_daemon(monkeypatch: pytes assert len(results) == 1 assert results[0].success is True assert results[0].repaired_count == 2 + + +# polylogue-t93b: the daemon whale pass. A component whose aggregate raw +# payload exceeds the ordinary fast-path envelope must not stay +# permanently resource-blocked when every member is stream-record-safe -- +# ``raw_materialization_whale_pass_candidate`` selects it for a dedicated, +# single-component pass at a widened envelope, and the very same +# ``repair_raw_materialization`` entrypoint (now ``raw_artifact_id``-scoped) +# converges it. Non-stream-safe oversized components must never be +# selected and must carry a distinct typed reason instead. + + +def test_raw_materialization_whale_pass_candidate_selects_stream_safe_blocked_component(tmp_path: Path) -> None: + from tests.infra.revision_backfill_benchmark import build_revision_chain_corpus + + raw_ids = build_revision_chain_corpus(tmp_path, superseded_count=9, final_payload_bytes=2_000) + config = _config(tmp_path) + + # Envelope wide enough to admit the whole chain: nothing is oversized, + # no escalation needed. + assert ( + repair_mod.raw_materialization_whale_pass_candidate( + config, ordinary_max_payload_bytes=10_000_000, whale_max_payload_bytes=10_000_000 + ) + is None + ) + + # Ordinary envelope too small for the chain, but the whale envelope is + # wide enough and every member is stream-safe (Provider.CODEX, .jsonl): + # the earliest-acquired raw (the fairness seed) is selected. + seed = repair_mod.raw_materialization_whale_pass_candidate( + config, ordinary_max_payload_bytes=500, whale_max_payload_bytes=10_000_000 + ) + assert seed == raw_ids[0] + + # Whale envelope also too small: genuinely still blocked, no candidate. + assert ( + repair_mod.raw_materialization_whale_pass_candidate( + config, ordinary_max_payload_bytes=500, whale_max_payload_bytes=500 + ) + is None + ) + + +def test_raw_materialization_whale_pass_candidate_excludes_non_stream_safe_component(tmp_path: Path) -> None: + from polylogue.core.enums import Provider + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + payload = json.dumps({"mapping": {}, "title": "non-stream-safe-whale"}).encode() + b"x" * 2_000 + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CHATGPT, + payload=payload, + source_path="chatgpt-export/conversations.json", + acquired_at_ms=1, + ) + config = _config(tmp_path) + + # Oversized under the ordinary envelope, within the whale envelope, but + # ChatGPT export + non-jsonl source path is not stream-record-safe -- + # must never be selected for escalation, at any whale envelope width. + assert ( + repair_mod.raw_materialization_whale_pass_candidate( + config, ordinary_max_payload_bytes=500, whale_max_payload_bytes=1_000_000_000 + ) + is None + ) + del raw_id + + +def test_raw_materialization_ordinary_pass_census_detail_distinguishes_escalation_eligibility( + tmp_path: Path, +) -> None: + """The durable census detail must say which of the two blocked reasons + applies, for one individually-oversized raw of each kind (Codex .jsonl + -- stream-safe -- versus ChatGPT .json -- not).""" + from polylogue.core.enums import Provider + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + stream_safe_raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b'{"type":"session_meta","payload":{"id":"t93b-stream-safe"}}\n', + source_path="codex/t93b-stream-safe.jsonl", + acquired_at_ms=1, + ) + non_stream_safe_raw_id = archive.write_raw_payload( + provider=Provider.CHATGPT, + payload=json.dumps({"mapping": {}, "title": "t93b-non-stream-safe"}).encode(), + source_path="chatgpt-export/conversations.json", + acquired_at_ms=2, + ) + ordinary_limit = 500 + with sqlite3.connect(tmp_path / "source.db") as conn: + conn.executemany( + "UPDATE raw_sessions SET blob_size = ? WHERE raw_id = ?", + ((ordinary_limit + 1, raw_id) for raw_id in (stream_safe_raw_id, non_stream_safe_raw_id)), + ) + conn.commit() + + stream_safe_result = repair_mod.repair_raw_materialization( + _config(tmp_path), raw_artifact_id=stream_safe_raw_id, max_payload_bytes=ordinary_limit + ) + non_stream_safe_result = repair_mod.repair_raw_materialization( + _config(tmp_path), raw_artifact_id=non_stream_safe_raw_id, max_payload_bytes=ordinary_limit + ) + + assert stream_safe_result.success is False + assert non_stream_safe_result.success is False + assert stream_safe_result.metrics.get("raw_materialization_stream_oversized_count", 0.0) >= 1.0 + assert non_stream_safe_result.metrics.get("raw_materialization_non_stream_safe_oversized_count", 0.0) >= 1.0 + + with sqlite3.connect(tmp_path / "source.db") as conn: + stream_safe_detail = conn.execute( + "SELECT detail FROM raw_authority_parser_census WHERE raw_id = ?", (stream_safe_raw_id,) + ).fetchone()[0] + non_stream_safe_detail = conn.execute( + "SELECT detail FROM raw_authority_parser_census WHERE raw_id = ?", (non_stream_safe_raw_id,) + ).fetchone()[0] + + assert "escalation-eligible: stream-safe" in stream_safe_detail + assert "escalation-blocked: non-stream-safe" in non_stream_safe_detail + + +def test_raw_materialization_whale_pass_converges_blocked_component_to_resolved_head(tmp_path: Path) -> None: + """The escalation-tier whale pass (raw_artifact_id-scoped, widened + envelope) must converge a component the ordinary envelope permanently + resource-blocks -- reusing the exact same convergence entrypoint, not a + parallel/offline code path. + + Anti-vacuity: removing the ``raw_artifact_id`` escalation scoping (i.e. + running the archive-wide ordinary pass at ``max_payload_bytes=whale_limit`` + without narrowing to one component -- simulated below by asserting the + ordinary-envelope call fails first) reproduces the permanently-blocked + bug this closes. + """ + from tests.infra.revision_backfill_benchmark import build_revision_chain_corpus + + raw_ids = build_revision_chain_corpus(tmp_path, superseded_count=9, final_payload_bytes=2_000) + config = _config(tmp_path) + ordinary_limit = 500 + whale_limit = 10_000_000 + + # The ordinary fast-path envelope permanently blocks the whole chain. + blocked = repair_mod.repair_raw_materialization( + config, raw_artifact_id=raw_ids[0], max_payload_bytes=ordinary_limit + ) + assert blocked.success is False + + seed = repair_mod.raw_materialization_whale_pass_candidate( + config, ordinary_max_payload_bytes=ordinary_limit, whale_max_payload_bytes=whale_limit + ) + assert seed is not None + + converged = repair_mod.repair_raw_materialization(config, raw_artifact_id=seed, max_payload_bytes=whale_limit) + + assert converged.success is True + assert converged.repaired_count >= 1 + with sqlite3.connect(tmp_path / "index.db") as conn: + row = conn.execute( + "SELECT native_id, origin FROM sessions WHERE native_id = ?", ("nh44-chain-session",) + ).fetchone() + assert row is not None + assert row[1] == "codex-session" + + # Fully converged: no longer a whale-pass candidate at any envelope. + assert ( + repair_mod.raw_materialization_whale_pass_candidate( + config, ordinary_max_payload_bytes=ordinary_limit, whale_max_payload_bytes=whale_limit + ) + is None + ) + + +def test_raw_materialization_whale_pass_commit_batches_bounded(tmp_path: Path) -> None: + """The whale pass must honor ``commit_batch_size`` -- shrinking the batch + must shrink writer-hold length within the escalated component (more, + smaller commit-boundary transactions), not commit the whole hundreds-raw + chain as one unbounded transaction regardless of the config knob. + + Anti-vacuity: if the whale-pass call path silently dropped + ``commit_batch_size`` (e.g. always resolving the single-big-batch + default no matter what the caller passed), ``fine_count`` and + ``coarse_count`` below would come out equal -- verified by asserting + they differ and that finer batching produces strictly more explicit + commits. + """ + import unittest.mock as mock + + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from tests.infra.revision_backfill_benchmark import build_revision_chain_corpus + + whale_limit = 10_000_000 + + def _run(root: Path, *, commit_batch_size: int) -> tuple[int, int]: + raw_ids = build_revision_chain_corpus(root, superseded_count=11, final_payload_bytes=2_000) + commit_count = 0 + original_commit = ArchiveStore.commit + + def counting_commit(self: ArchiveStore) -> None: + nonlocal commit_count + commit_count += 1 + original_commit(self) + + with mock.patch.object(ArchiveStore, "commit", counting_commit): + result = repair_mod.repair_raw_materialization( + _config(root), + raw_artifact_id=raw_ids[0], + max_payload_bytes=whale_limit, + commit_batch_size=commit_batch_size, + ) + assert result.success is True + return commit_count, len(raw_ids) + + fine_count, raw_count = _run(tmp_path / "fine", commit_batch_size=3) + coarse_count, _raw_count = _run(tmp_path / "coarse", commit_batch_size=1_000) + + assert coarse_count >= 1 + assert fine_count > coarse_count + assert fine_count <= raw_count