perf(sources): tree-byte decoded layer + in-cohort head-retire drift fix

polylogue/daemon/parse_prefetch.py

@@ -34,14 +34,17 @@
 
 import os
 import threading
-from collections.abc import Mapping, Sequence
+from collections.abc import Sequence
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from polylogue.config import Config
 from polylogue.logging import get_logger
+from polylogue.pipeline.parsed_tree_size import (
+    effective_physical_memory_bytes,
+    estimate_parsed_tree_bytes,
+)
 from polylogue.sources import revision_backfill
 from polylogue.sources.dispatch import is_stream_record_provider
-from polylogue.sources.parsers.base_models import ParsedSession
 from polylogue.sources.revision_backfill import RawParsePrefetchCache
 from polylogue.storage.repair import (
     raw_materialization_pending_census_raw_ids,
@@ -99,14 +102,7 @@ def daemon_parse_stage_worker_count() -> int:
 
 
 def _physical_memory_bytes() -> int | None:
-    try:
-        pages = os.sysconf("SC_PHYS_PAGES")
-        page_size = os.sysconf("SC_PAGE_SIZE")
-    except (ValueError, OSError, AttributeError):
-        return None
-    if pages <= 0 or page_size <= 0:
-        return None
-    return pages * page_size
+    return effective_physical_memory_bytes()
 
 
 def daemon_parse_stage_max_inflight_bytes() -> int:
@@ -149,126 +145,6 @@ def daemon_parse_stage_max_inflight_bytes() -> int:
 _MIN_MAX_CACHED_TREE_BYTES = 256 * 1024 * 1024  # 256 MiB
 _MAX_MAX_CACHED_TREE_BYTES = 4 * 1024 * 1024 * 1024  # 4 GiB
 
-# Calibration (measured 2026-07-20, see test_parse_prefetch.py for the exact
-# reproducer): a manual deep-object-graph walk (sys.getsizeof over every
-# reachable dict/list/model instance, the same technique pympler.asizeof
-# uses, without adding a new dependency for one calibration script) against
-# synthetic ParsedSession trees of increasing size gave:
-#
-#   messages=10,  blocks=20,   payload_chars=3_000     deep_bytes=41_165   (13.7x payload chars)
-#   messages=100, blocks=200,  payload_chars=150_000   deep_bytes=368_795  (2.5x payload chars)
-#   messages=500, blocks=1000, payload_chars=1_500_000 deep_bytes=2_521_995 (1.7x payload chars)
-#
-# A single per-char multiplier alone under-fits small/medium trees (fixed
-# per-object overhead dominates there) and a single per-object constant alone
-# under-fits text-heavy trees. A two-term linear fit (bytes_per_char * chars +
-# object_overhead_bytes * object_count, least-squares against the three
-# points above) recovers ~0.39 bytes/char and ~1290 bytes/object. This module
-# rounds BOTH terms up for a deliberate safety margin (favor overestimating
-# resident size, which biases toward eviction/reparse -- always correct --
-# over underestimating, which risks the exact OOM this budget exists to
-# prevent): 2 bytes/char and 1024 bytes/object, which lands within [0.9x,
-# 1.8x] of the measured deep size across the three calibration points.
-_ESTIMATOR_BYTES_PER_CHAR = 2
-_ESTIMATOR_OBJECT_OVERHEAD_BYTES = 1024
-
-
-def _text_len(value: str | None) -> int:
-    return len(value) if value else 0
-
-
-def _mapping_char_len(mapping: Mapping[str, object] | None) -> int:
-    """Cheap, non-recursive-getsizeof approximation of a dict-like field's size.
-
-    ``tool_input``/``metadata``/session-event ``payload`` fields are
-    provider-controlled dicts, occasionally large (a tool call's full JSON
-    args). A single ``repr()`` pass over each value is O(size) but touches
-    every byte, exactly the kind of per-object deep walk this estimator is
-    designed to avoid paying for the WHOLE tree -- so cap what a single
-    mapping field is allowed to contribute by falling back to ``str`` length
-    for string values (the overwhelmingly common case) and only ``repr``ing
-    non-string values, which are typically short (numbers, bools, small
-    nested structures).
-    """
-    if not mapping:
-        return 0
-    total = 0
-    for key, value in mapping.items():
-        total += len(str(key))
-        total += len(value) if isinstance(value, str) else len(repr(value))
-    return total
-
-
-def estimate_parsed_tree_bytes(sessions: Sequence[ParsedSession]) -> int:
-    """Cheap structural estimate of resident bytes for a parsed session tree.
-
-    Deliberately NOT a recursive ``sys.getsizeof``/pympler-style deep walk --
-    that is accurate but O(object graph size) with real per-call overhead,
-    and this runs on ``warm()``'s hot path for every raw in a page (up to a
-    couple thousand). Instead: a single linear pass sums text/content field
-    lengths and counts model-instance nodes (sessions, messages, blocks,
-    attachments, session events, web constructs), then applies two constants
-    calibrated against a real deep-size measurement -- see the constants'
-    docstring/comment above for the calibration data and measured ratio.
-    """
-    total_chars = 0
-    object_count = 0
-    for session in sessions:
-        object_count += 1
-        total_chars += _text_len(session.title)
-        total_chars += _text_len(session.instructions_text)
-        total_chars += _text_len(session.created_at)
-        total_chars += _text_len(session.updated_at)
-        total_chars += _text_len(session.git_branch)
-        total_chars += _text_len(session.git_repository_url)
-        total_chars += _text_len(session.provider_project_ref)
-        total_chars += _text_len(session.git_commit_hash)
-        total_chars += sum(len(value) for value in session.working_directories)
-        total_chars += sum(len(value) for value in session.models_used)
-        total_chars += sum(len(value) for value in session.ingest_flags)
-
-        for message in session.messages:
-            object_count += 1
-            object_count += len(message.paste_spans)
-            total_chars += _text_len(message.text)
-            total_chars += _text_len(message.timestamp)
-            total_chars += _text_len(message.sender_name)
-            total_chars += _text_len(message.recipient)
-            total_chars += _text_len(message.user_context_text)
-            total_chars += _text_len(message.model_name)
-            total_chars += _text_len(message.model_effort)
-            total_chars += _text_len(message.provider_message_id)
-            total_chars += _text_len(message.parent_message_provider_id)
-
-            for block in message.blocks:
-                object_count += 1
-                total_chars += _text_len(block.text)
-                total_chars += _text_len(block.tool_name)
-                total_chars += _text_len(block.tool_id)
-                total_chars += _text_len(block.media_type)
-                total_chars += _mapping_char_len(block.tool_input)
-                total_chars += _mapping_char_len(block.metadata)
-                for construct in block.web_constructs:
-                    object_count += 1
-                    total_chars += _text_len(construct.title)
-                    total_chars += _text_len(construct.text)
-                    total_chars += _text_len(construct.url)
-
-        for attachment in session.attachments:
-            object_count += 1
-            total_chars += _text_len(attachment.name)
-            total_chars += _text_len(attachment.path)
-            total_chars += _text_len(attachment.mime_type)
-            total_chars += _text_len(attachment.source_url)
-            total_chars += _text_len(attachment.caption)
-
-        for event in session.session_events:
-            object_count += 1
-            total_chars += _text_len(event.event_type)
-            total_chars += _mapping_char_len(event.payload)
-
-    return _ESTIMATOR_OBJECT_OVERHEAD_BYTES * object_count + _ESTIMATOR_BYTES_PER_CHAR * total_chars
-
 
 def daemon_parse_stage_max_cached_tree_bytes() -> int:
     """Whole-cache budget for ESTIMATED parsed-tree bytes (not payload bytes).