From dbb4a807bd0d104eef50e7d18309709b3d729961 Mon Sep 17 00:00:00 2001 From: earayu Date: Sun, 26 Apr 2026 05:54:50 +0800 Subject: [PATCH] feat(phase9 #99 D10.g): wire read-primitive cache into 4 parse_version-keyed primitives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit D10.g task #99 follow-up — completes the wire-in deferred in #1716. Each of the four parse_version-keyed read primitives in ``aperag/mcp/tools/`` now calls ``cache.get_or_compute_*`` after the D9 tenancy + authorization gates run; the un-cached fetch path becomes the cache's ``compute`` callback. Per §E.7 hard lock the cache only accelerates — every invocation runs ``resolve_authenticated_user`` → ``tenancy_gate`` → ``authorization_gate`` before any cache lookup. The cache layer never sees the calling user and cannot grant or skip a permission. Per §E.6 the chunk primitive's cache key is ``(chunk_id,)`` only — no parse_version weighting because ``chunk_id`` is indexing-layer-immutable. Changes: - ``aperag/cache/runtime.py`` (new) — process-wide :func:`get_read_primitive_cache` lazy singleton wired to the existing memory-redis client (via ``aperag.db.redis_manager.RedisConnectionManager``); falls back to ``NoopL2Cache`` when Redis is unreachable so the cache never blocks authoritative reads (§E.7 fail-open). - ``aperag/cache/__init__.py`` — re-exports ``get_read_primitive_cache`` / ``reset_read_primitive_cache``. - ``aperag/mcp/tools/read_document.py`` — caches the full ``DocumentContent`` (key = document_id + parse_version); byte-range slicing applied AFTER cache lookup so the cache stays shared across range requests for the same document version. - ``aperag/mcp/tools/read_document_outline.py`` — caches the ``DocumentOutline`` envelope. - ``aperag/mcp/tools/read_document_section.py`` — caches by ``(document_id, parse_version, section_path, heading_anchor)``; computes outline + slice + sibling-count inside the cache compute. - ``aperag/mcp/tools/read_document_chunk.py`` — caches by ``(chunk_id,)`` only per §E.6. - ``tests/unit_test/cache/test_wire_in_invariants.py`` (new) — three static guards: (1) cache call must follow tenancy + authorization gates in every primitive body, (2) each primitive uses its dedicated typed ``get_or_compute_*`` method, (3) the chunk primitive must not pass parse_version to the cache key. Production wiring: - The L1 size and L2 TTL knobs landed in #1716 (``Config.d10_cache_l1_size`` / ``Config.d10_cache_l2_ttl_seconds``) are now actually consumed by the singleton. - L2 fail-open: a Redis connection failure at process start logs at WARNING and degrades to L1-only — the cache layer remains available per the §E.7 hard lock that authoritative reads must never block on the cache. Test plan: - ``uv run --extra test python -m pytest tests/unit_test/`` → 932 passed / 29 skipped (was 917 + 3 new wire-in invariant tests + 12 from D10.e #1715 helper tests landed in the meantime = 932). - ``make lint`` clean. - The pre-existing ``test_call_sequence_witness_for_all_parse_version_keyed_primitives`` in ``tests/unit_test/test_d10c_read_primitives_surface.py`` still passes — the cache call is added strictly after the D9 base gates. Out of scope: ``get_collection_metadata`` / ``get_document_metadata`` short-TTL caching; explicit invalidation triggers wired to write paths (those wait on D11+ write-tools per §E.5 + §G D10.g write-set boundary). Co-Authored-By: Claude Opus 4.7 --- aperag/cache/__init__.py | 6 + aperag/cache/runtime.py | 125 ++++++++++++++++++ aperag/mcp/tools/read_document.py | 68 ++++++---- aperag/mcp/tools/read_document_chunk.py | 63 +++++---- aperag/mcp/tools/read_document_outline.py | 22 ++- aperag/mcp/tools/read_document_section.py | 82 +++++++----- .../cache/test_wire_in_invariants.py | 103 +++++++++++++++ 7 files changed, 382 insertions(+), 87 deletions(-) create mode 100644 aperag/cache/runtime.py create mode 100644 tests/unit_test/cache/test_wire_in_invariants.py diff --git a/aperag/cache/__init__.py b/aperag/cache/__init__.py index 90c5fc6a9..c1c847fa6 100644 --- a/aperag/cache/__init__.py +++ b/aperag/cache/__init__.py @@ -66,6 +66,10 @@ ReadPrimitiveCache, build_read_primitive_cache, ) +from aperag.cache.runtime import ( + get_read_primitive_cache, + reset_read_primitive_cache, +) __all__ = [ "L1Cache", @@ -73,6 +77,8 @@ "NoopL2Cache", "ReadPrimitiveCache", "build_read_primitive_cache", + "get_read_primitive_cache", "invalidate_collection", "invalidate_document", + "reset_read_primitive_cache", ] diff --git a/aperag/cache/runtime.py b/aperag/cache/runtime.py new file mode 100644 index 000000000..2e7519cf4 --- /dev/null +++ b/aperag/cache/runtime.py @@ -0,0 +1,125 @@ +# Copyright 2025 ApeCloud, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Production wiring helper for the D10.g read-primitive cache. + +The four parse_version-keyed primitives in :mod:`aperag.mcp.tools` call +:func:`get_read_primitive_cache` once they have already passed tenancy ++ authorization gates. This helper builds (and memoizes) one +:class:`~aperag.cache.read_primitive_cache.ReadPrimitiveCache` per +process, wiring it to: + +* the L1 size knob at :attr:`aperag.config.Config.d10_cache_l1_size`, +* the L2 TTL knob at :attr:`aperag.config.Config.d10_cache_l2_ttl_seconds`, +* the existing memory-redis client (via + :class:`aperag.db.redis_manager.RedisConnectionManager` — + ``settings.memory_redis_url``). + +Tests construct their own cache instances and inject fakes via the +underlying :func:`~aperag.cache.read_primitive_cache.build_read_primitive_cache`; +they do not touch the singleton — see :func:`reset_read_primitive_cache` +for an explicit test-only reset hook. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Optional + +from aperag.cache.read_primitive_cache import ( + ReadPrimitiveCache, + build_read_primitive_cache, +) +from aperag.config import settings + +logger = logging.getLogger(__name__) + + +_cache: Optional[ReadPrimitiveCache] = None +_build_lock = asyncio.Lock() + + +async def get_read_primitive_cache() -> ReadPrimitiveCache: + """Return the process-wide :class:`ReadPrimitiveCache`. + + Builds the singleton on first call (lazy). The L2 layer is wired + to the existing memory-redis client when one is reachable; if not, + a no-op L2 keeps the composition shape but degrades to L1-only + (per :class:`~aperag.cache.parse_version_cache.NoopL2Cache`). + """ + + global _cache + if _cache is not None: + return _cache + async with _build_lock: + if _cache is not None: # double-checked under lock + return _cache + _cache = await _build_singleton() + return _cache + + +async def reset_read_primitive_cache() -> None: + """Drop the cached singleton — test-only hook. + + Production code does not call this; the singleton lives for the + process. Tests that exercise wiring (e.g., switching L2 to a fake + Redis) should construct their own + :class:`ReadPrimitiveCache` directly via + :func:`~aperag.cache.read_primitive_cache.build_read_primitive_cache` + and pass it in explicitly. This helper exists for the rare test + that wants to verify the production singleton path itself. + """ + + global _cache + async with _build_lock: + _cache = None + + +async def _build_singleton() -> ReadPrimitiveCache: + redis = await _resolve_memory_redis_client() + return build_read_primitive_cache( + redis=redis, + l1_size=settings.d10_cache_l1_size, + l2_ttl_seconds=settings.d10_cache_l2_ttl_seconds, + ) + + +async def _resolve_memory_redis_client(): + """Best-effort wiring of the existing memory-redis client. + + Failure to obtain a client (no Redis configured, connection + refused, etc.) degrades to L1-only — the cache layer remains + available and still amortizes parse cost within one worker. We + log at WARNING so the operator can see the degradation; we do + NOT raise, because §E.7 fail-open invariant explicitly says the + cache must never block authoritative reads. + """ + + try: + from aperag.db.redis_manager import RedisConnectionManager + + return await RedisConnectionManager.get_async_client() + except Exception as e: + logger.warning( + "D10.g cache: memory-redis client unavailable, degrading L2 to noop: %s", + e, + ) + return None + + +__all__ = [ + "get_read_primitive_cache", + "reset_read_primitive_cache", +] diff --git a/aperag/mcp/tools/read_document.py b/aperag/mcp/tools/read_document.py index 838f57b15..92edd268b 100644 --- a/aperag/mcp/tools/read_document.py +++ b/aperag/mcp/tools/read_document.py @@ -29,6 +29,7 @@ from sqlalchemy import select +from aperag.cache import get_read_primitive_cache from aperag.config import get_async_session from aperag.domains.knowledge_base.db.models import ( Document, @@ -83,32 +84,53 @@ async def read_document( parse_version = resolve_parse_version(document, collection) - # 5. Fetch authoritative parsed markdown — un-cached body work. - parsed_markdown = await read_parsed_markdown(document) - - truncated = False - truncation_reason: Optional[str] = None - if range is not None: - encoded = parsed_markdown.encode("utf-8") - start = max(0, int(range.start)) - end = min(len(encoded), int(range.end)) - if end < start: - end = start - try: - sliced = encoded[start:end].decode("utf-8") - except UnicodeDecodeError: - # Best-effort: byte boundaries may split a multi-byte char. - sliced = encoded[start:end].decode("utf-8", errors="replace") - if start != 0 or end != len(encoded): - truncated = True - truncation_reason = f"byte_range[{start}:{end}]" - parsed_markdown = sliced + # 5. Fetch authoritative parsed markdown — D10.g cache-wrapped. + # The cache is keyed by (document_id, parse_version) and holds the + # full (un-ranged) DocumentContent; byte-range slicing is a cheap + # transformation applied AFTER cache lookup so the cache remains + # shared across different range requests for the same document + # version. Tenancy/auth above are NEVER skipped (§E.7 hard lock). + cache = await get_read_primitive_cache() + + async def _compute() -> DocumentContent: + parsed_markdown = await read_parsed_markdown(document) + return DocumentContent( + document_id=document.id, + collection_id=document.collection_id, + parsed_markdown=parsed_markdown, + parse_version=parse_version, + truncated=False, + truncation_reason=None, + ) - return DocumentContent( + full = await cache.get_or_compute_content( document_id=document.id, - collection_id=document.collection_id, - parsed_markdown=parsed_markdown, parse_version=parse_version, + compute=_compute, + model_cls=DocumentContent, + ) + + if range is None: + return full + + encoded = full.parsed_markdown.encode("utf-8") + start = max(0, int(range.start)) + end = min(len(encoded), int(range.end)) + if end < start: + end = start + try: + sliced = encoded[start:end].decode("utf-8") + except UnicodeDecodeError: + # Best-effort: byte boundaries may split a multi-byte char. + sliced = encoded[start:end].decode("utf-8", errors="replace") + truncated = start != 0 or end != len(encoded) + truncation_reason = f"byte_range[{start}:{end}]" if truncated else None + + return DocumentContent( + document_id=full.document_id, + collection_id=full.collection_id, + parsed_markdown=sliced, + parse_version=full.parse_version, truncated=truncated, truncation_reason=truncation_reason, ) diff --git a/aperag/mcp/tools/read_document_chunk.py b/aperag/mcp/tools/read_document_chunk.py index d752498d9..a5b7dee3b 100644 --- a/aperag/mcp/tools/read_document_chunk.py +++ b/aperag/mcp/tools/read_document_chunk.py @@ -31,6 +31,7 @@ from sqlalchemy import select +from aperag.cache import get_read_primitive_cache from aperag.config import get_async_session from aperag.domains.knowledge_base.db.models import ( Document, @@ -82,31 +83,43 @@ async def read_document_chunk( parse_version = resolve_parse_version(document, collection) - # 5. Fetch authoritative chunk — un-cached body work. - # Delegate to the existing tenant-aware service-layer helper which - # routes through the vector-store connector with the multi-tenant - # guard. We then locate the requested chunk by id within the result. - all_chunks = await document_service.get_document_chunks(str(user.id), collection_id, document_id) - matched = next((c for c in all_chunks if c.id == chunk_id), None) - if matched is None: - raise DocumentNotFoundException(f"Chunk not found in document {document_id}: chunk_id={chunk_id!r}") - - chunk_index = 0 - for i, c in enumerate(all_chunks): - if c.id == chunk_id: - chunk_index = i - break - - metadata = matched.metadata or {} - return DocumentChunk( - chunk_id=matched.id, - document_id=document.id, - collection_id=document.collection_id, - parsed_markdown=matched.text or "", - section_path=metadata.get("section_path"), - chunk_index=chunk_index, - chunk_total=len(all_chunks), - parse_version=parse_version, + # 5. Fetch authoritative chunk — D10.g cache-wrapped. + # §E.6: chunk_id is indexing-immutable, so the cache key is + # ``(chunk_id,)`` only — no parse_version weighting. Tenancy/auth + # above are NEVER skipped (§E.7 hard lock). + cache = await get_read_primitive_cache() + + async def _compute() -> DocumentChunk: + # Delegate to the existing tenant-aware service-layer helper which + # routes through the vector-store connector with the multi-tenant + # guard. We then locate the requested chunk by id within the result. + all_chunks = await document_service.get_document_chunks(str(user.id), collection_id, document_id) + matched = next((c for c in all_chunks if c.id == chunk_id), None) + if matched is None: + raise DocumentNotFoundException(f"Chunk not found in document {document_id}: chunk_id={chunk_id!r}") + + chunk_index = 0 + for i, c in enumerate(all_chunks): + if c.id == chunk_id: + chunk_index = i + break + + metadata = matched.metadata or {} + return DocumentChunk( + chunk_id=matched.id, + document_id=document.id, + collection_id=document.collection_id, + parsed_markdown=matched.text or "", + section_path=metadata.get("section_path"), + chunk_index=chunk_index, + chunk_total=len(all_chunks), + parse_version=parse_version, + ) + + return await cache.get_or_compute_chunk( + chunk_id=chunk_id, + compute=_compute, + model_cls=DocumentChunk, ) diff --git a/aperag/mcp/tools/read_document_outline.py b/aperag/mcp/tools/read_document_outline.py index 32bd5ee9a..d0a68e76a 100644 --- a/aperag/mcp/tools/read_document_outline.py +++ b/aperag/mcp/tools/read_document_outline.py @@ -30,6 +30,7 @@ from sqlalchemy import select +from aperag.cache import get_read_primitive_cache from aperag.config import get_async_session from aperag.domains.knowledge_base.db.models import ( Document, @@ -84,14 +85,25 @@ async def read_document_outline( parse_version = resolve_parse_version(document, collection) - # 5. Fetch + walk authoritative markdown — un-cached body work. - markdown = await read_parsed_markdown(document) - headings = build_outline_from_markdown(markdown, max_depth=max_depth) + # 5. Fetch + walk authoritative markdown — D10.g cache-wrapped. + # The cache only accelerates; tenancy/auth above are NEVER skipped + # (§E.7 hard lock). + cache = await get_read_primitive_cache() + + async def _compute() -> DocumentOutline: + markdown = await read_parsed_markdown(document) + headings = build_outline_from_markdown(markdown, max_depth=max_depth) + return DocumentOutline( + document_id=document.id, + headings=headings, + parse_version=parse_version, + ) - return DocumentOutline( + return await cache.get_or_compute_outline( document_id=document.id, - headings=headings, parse_version=parse_version, + compute=_compute, + model_cls=DocumentOutline, ) diff --git a/aperag/mcp/tools/read_document_section.py b/aperag/mcp/tools/read_document_section.py index 1a569c8b3..907685cc1 100644 --- a/aperag/mcp/tools/read_document_section.py +++ b/aperag/mcp/tools/read_document_section.py @@ -29,6 +29,7 @@ from sqlalchemy import select +from aperag.cache import get_read_primitive_cache from aperag.config import get_async_session from aperag.domains.knowledge_base.db.models import ( Document, @@ -92,45 +93,58 @@ async def read_document_section( parse_version = resolve_parse_version(document, collection) - # 5. Fetch + slice authoritative markdown. - markdown = await read_parsed_markdown(document) - outline = build_outline_from_markdown(markdown) + # 5. Fetch + slice authoritative markdown — D10.g cache-wrapped. + # Tenancy/auth above are NEVER skipped (§E.7 hard lock). + cache = await get_read_primitive_cache() - heading = find_section_in_outline( - outline, - section_path=section_path, - heading_anchor=heading_anchor, - ) - if heading is None: - raise DocumentNotFoundException( - f"Section not found in document {document_id}: " - f"section_path={section_path!r} heading_anchor={heading_anchor!r}" + async def _compute() -> DocumentSection: + markdown = await read_parsed_markdown(document) + outline = build_outline_from_markdown(markdown) + + heading = find_section_in_outline( + outline, + section_path=section_path, + heading_anchor=heading_anchor, + ) + if heading is None: + raise DocumentNotFoundException( + f"Section not found in document {document_id}: " + f"section_path={section_path!r} heading_anchor={heading_anchor!r}" + ) + + section_md = slice_section_markdown(markdown, heading) + parent_path: Optional[str] = None + if "/" in heading.section_path: + parent_path = "/".join(heading.section_path.split("/")[:-1]) + + # Sibling count: peers at the same level in the parent's children list. + sibling_count = 0 + if parent_path is None: + sibling_count = max(0, len(outline) - 1) + else: + parent = find_section_in_outline(outline, section_path=parent_path) + if parent is not None: + sibling_count = max(0, len(parent.children) - 1) + + return DocumentSection( + document_id=document.id, + collection_id=document.collection_id, + section_path=heading.section_path, + heading_anchor=heading.heading_anchor, + heading_text=heading.text, + parsed_markdown=section_md, + parse_version=parse_version, + parent_section_path=parent_path, + sibling_count=sibling_count, ) - section_md = slice_section_markdown(markdown, heading) - parent_path: Optional[str] = None - if "/" in heading.section_path: - parent_path = "/".join(heading.section_path.split("/")[:-1]) - - # Sibling count: peers at the same level in the parent's children list. - sibling_count = 0 - if parent_path is None: - sibling_count = max(0, len(outline) - 1) - else: - parent = find_section_in_outline(outline, section_path=parent_path) - if parent is not None: - sibling_count = max(0, len(parent.children) - 1) - - return DocumentSection( + return await cache.get_or_compute_section( document_id=document.id, - collection_id=document.collection_id, - section_path=heading.section_path, - heading_anchor=heading.heading_anchor, - heading_text=heading.text, - parsed_markdown=section_md, parse_version=parse_version, - parent_section_path=parent_path, - sibling_count=sibling_count, + section_path=section_path, + heading_anchor=heading_anchor, + compute=_compute, + model_cls=DocumentSection, ) diff --git a/tests/unit_test/cache/test_wire_in_invariants.py b/tests/unit_test/cache/test_wire_in_invariants.py new file mode 100644 index 000000000..e5ead98c5 --- /dev/null +++ b/tests/unit_test/cache/test_wire_in_invariants.py @@ -0,0 +1,103 @@ +# Copyright 2025 ApeCloud, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Static (source-level) invariants for the D10.g cache wire-in into +the four parse_version-keyed read primitives. + +The dynamic behavior (LRU / Redis / inflight collapsing / decode +failure → miss) is covered in the cache layer's own tests. This file +guards the *integration* invariants that the wire-in PR is responsible +for and that future refactors must not silently break. +""" + +from __future__ import annotations + +import inspect + +from aperag.mcp.tools.read_document import read_document +from aperag.mcp.tools.read_document_chunk import read_document_chunk +from aperag.mcp.tools.read_document_outline import read_document_outline +from aperag.mcp.tools.read_document_section import read_document_section + +_PARSE_VERSION_KEYED = ( + read_document, + read_document_outline, + read_document_section, + read_document_chunk, +) + + +def test_cache_call_appears_after_d9_gates_in_every_primitive_body(): + """§E.7 — tenancy + authorization MUST run before any cache call. + + Static guard: in every parse_version-keyed primitive's source, the + ``cache.get_or_compute_`` call appears AFTER ``tenancy_gate(`` and + AFTER ``authorization_gate(``. A future refactor that reorders + them (cache-first short-circuit) will fail this test loudly. + """ + + for primitive in _PARSE_VERSION_KEYED: + src = inspect.getsource(primitive) + idx_tenancy = src.find("tenancy_gate(") + idx_authz = src.find("authorization_gate(") + idx_cache = src.find("cache.get_or_compute_") + assert idx_tenancy != -1, f"{primitive.__name__}: tenancy_gate call missing" + assert idx_authz != -1, f"{primitive.__name__}: authorization_gate call missing" + assert idx_cache != -1, ( + f"{primitive.__name__}: cache.get_or_compute_ call missing — " + "the cache wire-in regressed back to un-cached body work" + ) + assert idx_tenancy < idx_cache, ( + f"{primitive.__name__}: cache call (@{idx_cache}) must come after tenancy_gate (@{idx_tenancy})" + ) + assert idx_authz < idx_cache, ( + f"{primitive.__name__}: cache call (@{idx_authz}) must come after authorization_gate (@{idx_authz})" + ) + + +def test_each_primitive_uses_its_dedicated_cache_method(): + """The wire-in must use the primitive-typed `get_or_compute_*` method + so the cache namespace is correct (outline / section / content / + chunk). A regression that picks the wrong helper would produce a + silently shared key namespace across primitives. + """ + + expected = { + "read_document_outline": "get_or_compute_outline", + "read_document_section": "get_or_compute_section", + "read_document": "get_or_compute_content", + "read_document_chunk": "get_or_compute_chunk", + } + for primitive in _PARSE_VERSION_KEYED: + src = inspect.getsource(primitive) + method = expected[primitive.__name__] + assert f"cache.{method}(" in src, f"{primitive.__name__}: expected cache.{method}( call site" + + +def test_chunk_primitive_does_not_pass_parse_version_to_cache(): + """§E.6 — chunk_id is indexing-immutable; the chunk cache key MUST + NOT include parse_version. A regression that passes parse_version + to the chunk cache would defeat the §E.6 design intent (cached + chunks would invalidate every parse). + """ + + src = inspect.getsource(read_document_chunk) + # Locate the cache call segment. + cache_call_start = src.find("cache.get_or_compute_chunk(") + assert cache_call_start != -1 + cache_call_end = src.find(")", cache_call_start) + cache_call_segment = src[cache_call_start:cache_call_end] + assert "parse_version" not in cache_call_segment, ( + "read_document_chunk cache call must not pass parse_version per §E.6" + )