feat(daemon): converge whale raw components via bounded streaming escalation pass (polylogue-t93b)

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. |

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,

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.