diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 8864ee6a1d..5ea0a3bb52 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -29,15 +29,15 @@ files: target: polylogue/agent_integration/data/__init__.py owner: stable - path: polylogue/agent_integration/installer.py - loc: 1348 + loc: 1367 target: polylogue/agent_integration/installer.py owner: stable - path: polylogue/agent_integration/manifest.py - loc: 132 + loc: 128 target: polylogue/agent_integration/manifest.py owner: stable - path: polylogue/agent_integration/spec.py - loc: 951 + loc: 945 target: polylogue/agent_integration/spec.py owner: stable - path: polylogue/annotations/__init__.py @@ -820,7 +820,7 @@ files: target: polylogue/cli/commands/__init__.py owner: stable - path: polylogue/cli/commands/agent.py - loc: 195 + loc: 216 target: polylogue/cli/commands/agent.py owner: stable - path: polylogue/cli/commands/agents.py @@ -1270,7 +1270,7 @@ files: target: polylogue/cli/verb_names.py owner: stable - path: polylogue/config.py - loc: 2367 + loc: 2501 target: polylogue/config.py owner: kernel reason: kernel root rule @@ -1645,7 +1645,7 @@ files: target: polylogue/daemon/otlp_receiver.py owner: stable - path: polylogue/daemon/parse_prefetch.py - loc: 280 + loc: 433 target: polylogue/daemon/parse_prefetch.py owner: stable - path: polylogue/daemon/process_start.py @@ -2111,7 +2111,7 @@ files: target: polylogue/maintenance/registry.py owner: stable - path: polylogue/maintenance/replay.py - loc: 845 + loc: 855 target: polylogue/maintenance/replay.py owner: stable - path: polylogue/maintenance/scope.py @@ -2187,23 +2187,23 @@ files: target: polylogue/mcp/call_log.py owner: stable - path: polylogue/mcp/cli.py - loc: 34 + loc: 42 target: polylogue/mcp/cli.py owner: stable - path: polylogue/mcp/declarations/__init__.py - loc: 57 + loc: 59 target: polylogue/mcp/declarations/__init__.py owner: stable - path: polylogue/mcp/declarations/adapter.py - loc: 175 + loc: 176 target: polylogue/mcp/declarations/adapter.py owner: stable - path: polylogue/mcp/declarations/models.py - loc: 225 + loc: 246 target: polylogue/mcp/declarations/models.py owner: stable - path: polylogue/mcp/declarations/registry.py - loc: 528 + loc: 539 target: polylogue/mcp/declarations/registry.py owner: stable - path: polylogue/mcp/insight_tool_contracts.py @@ -2223,11 +2223,11 @@ files: target: polylogue/mcp/query_contracts.py owner: stable - path: polylogue/mcp/server.py - loc: 117 + loc: 124 target: polylogue/mcp/server.py owner: stable - path: polylogue/mcp/server_cutover.py - loc: 1925 + loc: 1927 target: polylogue/mcp/server_cutover.py owner: stable - path: polylogue/mcp/server_prompts.py @@ -2235,11 +2235,11 @@ files: target: polylogue/mcp/server_prompts.py owner: stable - path: polylogue/mcp/server_resources.py - loc: 423 + loc: 420 target: polylogue/mcp/server_resources.py owner: stable - path: polylogue/mcp/server_support.py - loc: 698 + loc: 691 target: polylogue/mcp/server_support.py owner: stable - path: polylogue/mcp/server_tools.py @@ -2322,6 +2322,10 @@ files: loc: 70 target: polylogue/pipeline/ingest_support.py owner: stable + - path: polylogue/pipeline/parsed_tree_size.py + loc: 137 + target: polylogue/pipeline/parsed_tree_size.py + owner: stable - path: polylogue/pipeline/payload_types.py loc: 142 target: polylogue/pipeline/payload_types.py @@ -3100,7 +3104,7 @@ files: target: polylogue/sources/decoders.py owner: stable - path: polylogue/sources/dispatch.py - loc: 1097 + loc: 1102 target: polylogue/sources/dispatch.py owner: stable - path: polylogue/sources/drive/__init__.py @@ -3228,7 +3232,7 @@ files: target: polylogue/sources/live/watcher.py owner: stable - path: polylogue/sources/origin_specs.py - loc: 859 + loc: 868 target: polylogue/sources/origin_specs.py owner: stable - path: polylogue/sources/parsers/antigravity.py @@ -3393,7 +3397,7 @@ files: owner: stable cross_cut: { lifecycle: model } - path: polylogue/sources/revision_backfill.py - loc: 1402 + loc: 1512 target: polylogue/sources/revision_backfill.py owner: stable - path: polylogue/sources/source_acquisition.py @@ -3961,7 +3965,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/__init__.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive.py - loc: 12471 + loc: 12564 target: polylogue/storage/sqlite/archive_tiers/archive.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/archive_init.py @@ -4001,7 +4005,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/embeddings.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index.py - loc: 1633 + loc: 1658 target: polylogue/storage/sqlite/archive_tiers/index.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/index_convergence.py @@ -4021,7 +4025,7 @@ files: target: polylogue/storage/sqlite/archive_tiers/ops_write.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/pricing_seed.py - loc: 143 + loc: 157 target: polylogue/storage/sqlite/archive_tiers/pricing_seed.py owner: stable - path: polylogue/storage/sqlite/archive_tiers/revision_application.py @@ -4352,7 +4356,7 @@ files: target: polylogue/storage/sqlite/run_projection_relations.py owner: stable - path: polylogue/storage/sqlite/runtime_indexes.py - loc: 59 + loc: 68 target: polylogue/storage/sqlite/runtime_indexes.py owner: stable - path: polylogue/storage/sqlite/schema.py diff --git a/docs/topology-status.md b/docs/topology-status.md index 193847ed2d..32a70d09fa 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): 893 +- **Stable** (no move scoped): 894 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 9 -- **Total declared**: 1064 -- **Realized polylogue/**/*.py**: 1064 files declared +- **Total declared**: 1065 +- **Realized polylogue/**/*.py**: 1065 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/daemon/parse_prefetch.py b/polylogue/daemon/parse_prefetch.py index 92af6fec92..73edc4f01e 100644 --- a/polylogue/daemon/parse_prefetch.py +++ b/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). diff --git a/polylogue/pipeline/parsed_tree_size.py b/polylogue/pipeline/parsed_tree_size.py new file mode 100644 index 0000000000..7073d8c86b --- /dev/null +++ b/polylogue/pipeline/parsed_tree_size.py @@ -0,0 +1,183 @@ +"""Cheap structural size estimation for parsed session trees. + +Shared by the daemon parse-prefetch cache (polylogue-xb4i) and the +historical-backfill census spill's decoded layer: any component that retains +``ParsedSession`` trees in RAM budgets them by ESTIMATED TREE BYTES, never by +raw payload bytes (parsed trees inflate payload size by roughly 2-14x +depending on text density -- see the calibration data below). +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping, Sequence +from pathlib import Path + +from polylogue.sources.parsers.base import ParsedSession + +# 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 + total_inline_attachment_bytes = 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) + if attachment.inline_bytes is not None: + # Embedded attachment payloads stay resident in the decoded + # session (and in the spill pickle) byte-for-byte -- count + # them directly, no char multiplier. + total_inline_attachment_bytes += len(attachment.inline_bytes) + + 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 + + total_inline_attachment_bytes + ) + + +def effective_physical_memory_bytes() -> int | None: + """Physical RAM available to THIS process: host RAM capped by cgroup limit. + + Under a container/systemd cgroup memory limit, ``SC_PHYS_PAGES`` reports + host RAM; sizing an adaptive cache from it starves a memory-limited + process. Reads cgroup v2 ``memory.max`` (and legacy v1 + ``memory.limit_in_bytes``) and returns the minimum of host RAM and any + finite limit; ``None`` when neither is knowable. + """ + + physical: int | None + try: + pages = os.sysconf("SC_PHYS_PAGES") + page_size = os.sysconf("SC_PAGE_SIZE") + physical = pages * page_size if pages > 0 and page_size > 0 else None + except (ValueError, OSError, AttributeError): + physical = None + limits = [] + for path in ("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"): + try: + raw = Path(path).read_text().strip() + except OSError: + continue + if raw and raw != "max" and raw.isdigit(): + value = int(raw) + # v1 reports an enormous sentinel when unlimited + if 0 < value < (1 << 60): + limits.append(value) + if limits: + smallest = min(limits) + return smallest if physical is None else min(physical, smallest) + return physical + + +__all__ = ["estimate_parsed_tree_bytes", "effective_physical_memory_bytes"] diff --git a/polylogue/sources/revision_backfill.py b/polylogue/sources/revision_backfill.py index 88a7197514..66e72bfdb6 100644 --- a/polylogue/sources/revision_backfill.py +++ b/polylogue/sources/revision_backfill.py @@ -32,6 +32,7 @@ from polylogue.archive.session_revision_membership import MembershipRevision, classify_membership_revisions from polylogue.core.enums import Provider from polylogue.pipeline.ids import session_revision_projection +from polylogue.pipeline.parsed_tree_size import effective_physical_memory_bytes, estimate_parsed_tree_bytes from polylogue.pipeline.services.process_pool import parallel_threads_effective from polylogue.sources.decoders import _iter_json_stream from polylogue.sources.dispatch import is_stream_record_provider, parse_payload, parse_stream_payload @@ -1312,7 +1313,14 @@ class _ParsedSessionSpill: #: consumes a cohort almost immediately after census parses it, so a #: small hot layer turns the common for_raw() into a dict hit instead of #: a pickle.loads round-trip. Bounded independently of the sqlite layer. - _DECODED_CACHE_PAYLOAD_BYTES: Final[int] = 256 * 1024 * 1024 + #: Decoded-session RAM layer budget, accounted in ESTIMATED TREE BYTES + #: (polylogue-xb4i estimator) rather than raw payload bytes -- parsed + #: trees inflate payload 2-14x, so a payload-denominated budget either + #: wastes RAM headroom on text-dense sessions or overshoots on + #: structure-dense ones. Adaptive: physical RAM / 16, clamped to + #: [256 MiB, 2 GiB]; falls back to the floor when RAM is unknown. + _DECODED_CACHE_MIN_TREE_BYTES: Final[int] = 256 * 1024 * 1024 + _DECODED_CACHE_MAX_TREE_BYTES: Final[int] = 2 * 1024 * 1024 * 1024 def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None) -> None: # Place the spill beside the RESOLVED index tier, not the archive @@ -1347,8 +1355,14 @@ def __init__(self, archive_root: Path, *, max_cached_payload_bytes: int | None) self.conn.execute("CREATE INDEX parsed_sessions_logical ON parsed_sessions(logical_key, raw_id)") self.max_cached_payload_bytes = max_cached_payload_bytes self.cached_payload_bytes = 0 - self._decoded: dict[str, tuple[list[ParsedSession], int]] = {} - self._decoded_payload_bytes = 0 + self._decoded: dict[str, tuple[list[ParsedSession], int, int]] = {} + self._decoded_tree_bytes = 0 + physical = effective_physical_memory_bytes() or 0 + self._decoded_budget = ( + max(self._DECODED_CACHE_MIN_TREE_BYTES, min(self._DECODED_CACHE_MAX_TREE_BYTES, physical // 16)) + if physical + else self._DECODED_CACHE_MIN_TREE_BYTES + ) def __enter__(self) -> _ParsedSessionSpill: return self @@ -1385,19 +1399,20 @@ def add(self, raw_id: str, sessions: list[ParsedSession], *, payload_bytes: int) self._retain_decoded(raw_id, sessions, payload_bytes=payload_bytes) def _retain_decoded(self, raw_id: str, sessions: list[ParsedSession], *, payload_bytes: int) -> None: - if payload_bytes > self._DECODED_CACHE_PAYLOAD_BYTES: + tree_bytes = estimate_parsed_tree_bytes(sessions) + if tree_bytes > self._decoded_budget: return - while self._decoded and self._decoded_payload_bytes + payload_bytes > self._DECODED_CACHE_PAYLOAD_BYTES: + while self._decoded and self._decoded_tree_bytes + tree_bytes > self._decoded_budget: oldest_raw = next(iter(self._decoded)) - _evicted_sessions, evicted_bytes = self._decoded.pop(oldest_raw) - self._decoded_payload_bytes -= evicted_bytes - self._decoded[raw_id] = (sessions, payload_bytes) - self._decoded_payload_bytes += payload_bytes + _sessions, _payload, evicted_tree = self._decoded.pop(oldest_raw) + self._decoded_tree_bytes -= evicted_tree + self._decoded[raw_id] = (sessions, payload_bytes, tree_bytes) + self._decoded_tree_bytes += tree_bytes def for_raw(self, archive: ArchiveStore, raw_id: str) -> tuple[list[ParsedSession], int]: decoded = self._decoded.get(raw_id) if decoded is not None: - return decoded + return decoded[0], decoded[1] rows = self.conn.execute( "SELECT parsed, payload_bytes FROM parsed_sessions WHERE raw_id = ? ORDER BY logical_key", (raw_id,) ).fetchall() diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 7836cb4e67..9a4c653510 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -3262,12 +3262,25 @@ def apply_raw_membership_classification( "SELECT raw_id, content_hash FROM sessions WHERE session_id = ?", (str(existing_head[3]),), ).fetchone() + # Retiring an IN-COHORT head only requires that the + # persisted session row was also written by this + # cohort (head raw or any classified member): both + # representative drift across resumed passes (cohort + # absorption can flip which equivalent member wrote + # the session row) and content-hash drift (parser + # fixes between resumed passes re-derive hashes; the + # same-raw CAS exemption already treats that as + # re-derivation, not conflict) are healed immediately + # by this very replay re-indexing the accepted + # member. Only a persisted session written by a raw + # FOREIGN to the cohort still refuses -- that is the + # genuine unrelated-head hazard this guard exists + # for. + persisted_raw = None if persisted_session is None else str(persisted_session[0]) if ( existing_raw_id not in classified_raw_ids or persisted_session is None - or str(persisted_session[0]) != existing_raw_id - or not isinstance(persisted_session[1], bytes) - or bytes(existing_head[1]) != bytes(persisted_session[1]) + or (persisted_raw != existing_raw_id and persisted_raw not in classified_raw_ids) ): raise RuntimeError( "membership replay cannot retire an unrelated accepted head: " @@ -3279,12 +3292,17 @@ def apply_raw_membership_classification( f"ambiguous={classification.ambiguous_raw_ids!r}) " f"persisted_session_raw={None if persisted_session is None else str(persisted_session[0])!r}" ) - existing_is_byte_governed = conn.execute( - "SELECT 1 FROM raw_sessions WHERE raw_id = ? AND logical_source_key = ?", - (existing_raw_id, logical_source_key), - ).fetchone() - if existing_is_byte_governed is not None and accepted_raw_id != existing_raw_id: - raise RuntimeError("membership replay cannot replace an unconvertible byte head") + # No byte-governance refusal here: this branch is only + # reachable when the head raw is a member of THIS + # cohort, and membership rows for a source-keyed raw + # only exist after a governance conversion decided it + # belongs to membership classification -- a still-set + # logical_source_key on raw_sessions is interrupted- + # pass ordering drift (the conversion's key-nulling + # had not committed when the pass died), not evidence + # of live byte-chain governance. Foreign byte heads + # never reach this branch: they yield in the + # chain-governed-head branch above. self._conn.execute( "DELETE FROM raw_revision_heads WHERE logical_source_key = ?", (logical_source_key,),