feat(phase9 #99 D10.g): wire read-primitive cache into 4 parse_version-keyed primitives

aperag/cache/__init__.py

@@ -66,13 +66,19 @@
     ReadPrimitiveCache,
     build_read_primitive_cache,
 )
+from aperag.cache.runtime import (
+    get_read_primitive_cache,
+    reset_read_primitive_cache,
+)
 
 __all__ = [
     "L1Cache",
     "L2Cache",
     "NoopL2Cache",
     "ReadPrimitiveCache",
     "build_read_primitive_cache",
+    "get_read_primitive_cache",
     "invalidate_collection",
     "invalidate_document",
+    "reset_read_primitive_cache",
 ]

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",
+]

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,
     )

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,
     )
 
 

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,
     )