From 9fa61b72475e3e6f1ab5a664e678a7d90021ca93 Mon Sep 17 00:00:00 2001 From: phernandez Date: Thu, 13 Aug 2026 13:26:06 -0500 Subject: [PATCH] perf(mcp): optimize semantic note reads Signed-off-by: phernandez --- docs/REDIS_READ_CACHE_PLAN.md | 37 +- src/basic_memory/mcp/clients/knowledge.py | 8 +- src/basic_memory/mcp/note_reads.py | 106 ++++++ src/basic_memory/mcp/tools/read_note.py | 153 ++++---- src/basic_memory/read_cache/contract.py | 3 + src/basic_memory/read_cache/invalidation.py | 2 +- src/basic_memory/read_cache/policy.py | 2 +- src/basic_memory/read_cache/read_through.py | 79 ++++- src/basic_memory/read_cache/redis.py | 21 +- test-int/read_cache/test_redis_read_cache.py | 21 +- tests/mcp/test_read_note_request_counts.py | 350 +++++++++++++++++++ tests/mcp/test_tool_read_note.py | 8 +- tests/mcp/test_tool_telemetry.py | 2 +- tests/test_read_cache_telemetry.py | 323 +++++++++++++++++ 14 files changed, 996 insertions(+), 119 deletions(-) create mode 100644 src/basic_memory/mcp/note_reads.py create mode 100644 tests/mcp/test_read_note_request_counts.py create mode 100644 tests/test_read_cache_telemetry.py diff --git a/docs/REDIS_READ_CACHE_PLAN.md b/docs/REDIS_READ_CACHE_PLAN.md index 8e4610843..5a62d5379 100644 --- a/docs/REDIS_READ_CACHE_PLAN.md +++ b/docs/REDIS_READ_CACHE_PLAN.md @@ -226,25 +226,25 @@ Random tokens prevent an evicted generation key from returning to an old integer reviving stale data. A read that fills after concurrent invalidation also remains safe because its old token no longer matches. -## Initial Cache Surface +## Cache Surface And Production TTL Phase one: -| Operation | Initial TTL | Constraints | -| --------------------------- | ----------: | --------------------------------------------------- | -| Entity by external ID | 60 seconds | Cache validated `EntityResponseV2` JSON | -| Identifier resolution | 60 seconds | Include body and workspace context | -| Markdown note resource | 60 seconds | Cache only below an explicit size limit | -| Directory structure | 60 seconds | Folder-only tree; two MiB payload cap | -| Directory tree | 60 seconds | Full hierarchy; two MiB measured payload cap | -| Paginated directory listing | 60 seconds | Include path, depth, glob, page, and page-size keys | +| Operation | Production TTL | Constraints | +| --------------------------- | -------------: | --------------------------------------------------- | +| Entity by external ID | 300 seconds | Cache validated `EntityResponseV2` JSON | +| Identifier resolution | 300 seconds | Include body and workspace context | +| Markdown note resource | 300 seconds | Cache only below an explicit size limit | +| Directory structure | 300 seconds | Folder-only tree; two MiB payload cap | +| Directory tree | 300 seconds | Full hierarchy; two MiB measured payload cap | +| Paginated directory listing | 300 seconds | Include path, depth, glob, page, and page-size keys | Additional measured surfaces: -| Operation | Initial TTL | Constraints | -| --------------------------- | ------------: | ---------------------------------------------- | -| Search | 30 seconds | Implemented; complete query plus pagination key | -| Context and recent activity | 15-30 seconds | Future; normalize or bound time-relative inputs | +| Operation | Production TTL | Constraints | +| --------------------------- | -------------: | ----------------------------------------------- | +| Search | 30 seconds | Implemented; complete query plus pagination key | +| Context and recent activity | 15-30 seconds | Future; normalize or bound time-relative inputs | Do not initially cache failures, missing entities, graph/orphan responses, large or arbitrary binary resources, schema inference, writes, or Cloud control-plane data. @@ -437,7 +437,8 @@ response-cache layers as the final design. - Reads bypass Redis and use the authoritative path when the cache is unavailable. - Cache-store failures do not fail an otherwise successful read. - Cache-invalidation failures do not fail committed writes, but they emit prominent telemetry. -- Short initial TTLs bound stale-data exposure after an invalidation failure and Redis recovery. +- Bounded operation TTLs limit stale-data exposure after an invalidation failure and Redis + recovery. - Redis client-input errors plus local serialization, decoding, and programming errors fail fast rather than masquerading as cache misses. @@ -447,12 +448,14 @@ Rate-limit failure behavior remains entirely Cloud-owned. Record: -- hit, miss, bypass, store, invalidation, unavailable, and oversize outcomes; -- operation name without tenant or project metric labels; +- distinct lookup and store outcomes, including hit, miss, bypass, store, invalidation, + unavailable, corrupt, and oversize; +- operation name and configured TTL without tenant or project metric labels; +- remaining TTL on cache hits; - Redis operation latency; - cached payload size; - authoritative read latency on misses; -- hashed scope and request identifiers on diagnostic spans only. +- hashed scope, request, and generation identifiers on diagnostic spans only. Do not add public cache headers in the first version. diff --git a/src/basic_memory/mcp/clients/knowledge.py b/src/basic_memory/mcp/clients/knowledge.py index d4a8c1d0a..d3d0d6ee4 100644 --- a/src/basic_memory/mcp/clients/knowledge.py +++ b/src/basic_memory/mcp/clients/knowledge.py @@ -19,7 +19,7 @@ DirectoryDeleteResult, ) from basic_memory.schemas.v2.graph import GraphNode, OrphanEntitiesResponse -from basic_memory.schemas.v2.entity import EntityResolveResponse +from basic_memory.schemas.v2.entity import EntityResolveResponse, EntityResponseV2 class KnowledgeClient: @@ -112,14 +112,14 @@ async def update_entity( ) return EntityResponse.model_validate(response.json()) - async def get_entity(self, entity_id: str) -> EntityResponse: + async def get_entity(self, entity_id: str) -> EntityResponseV2: """Get an entity by ID. Args: entity_id: Entity external_id (UUID) Returns: - EntityResponse with entity details + EntityResponseV2 with accepted note content and entity metadata Raises: ToolError: If the entity is not found or request fails @@ -138,7 +138,7 @@ async def get_entity(self, entity_id: str) -> EntityResponse: operation="get_entity", path_template="/v2/projects/{project_id}/knowledge/entities/{entity_id}", ) - return EntityResponse.model_validate(response.json()) + return EntityResponseV2.model_validate(response.json()) async def patch_entity( self, diff --git a/src/basic_memory/mcp/note_reads.py b/src/basic_memory/mcp/note_reads.py new file mode 100644 index 000000000..6146891ec --- /dev/null +++ b/src/basic_memory/mcp/note_reads.py @@ -0,0 +1,106 @@ +"""Reusable typed note-read shaping for MCP adapters.""" + +from typing import Any, Protocol, TypedDict + +import logfire +import yaml +from httpx import Response + +from basic_memory.schemas.v2 import EntityResponseV2 + + +class ReadNoteJsonPayload(TypedDict): + """Existing successful ``read_note(output_format="json")`` payload.""" + + title: str + permalink: str | None + file_path: str + content: str + frontmatter: dict[str, Any] | None + + +class KnowledgeEntityReader(Protocol): + """Entity-read capability required by exact-ID note reads.""" + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + """Return the entity response for one exact external ID.""" + + +class NoteResourceReader(Protocol): + """Resource-read capability used only for legacy entities without content.""" + + async def read(self, entity_id: str) -> Response: + """Return raw resource content for one exact external ID.""" + + +def parse_opening_frontmatter(content: str) -> tuple[str, dict[str, Any] | None]: + """Parse opening YAML frontmatter and return ``(body, frontmatter)``. + + Mirrors CLI behavior: only parses a frontmatter block at the very top. + If parsing fails or frontmatter is not a mapping, returns body unchanged and ``None``. + """ + original_content = content + lines = content.splitlines(keepends=True) + if not lines or lines[0].strip() != "---": + return original_content, None + + closing_index = None + for index in range(1, len(lines)): + if lines[index].strip() == "---": + closing_index = index + break + + if closing_index is None: + return original_content, None + + frontmatter_text = "".join(lines[1:closing_index]) + try: + parsed = yaml.safe_load(frontmatter_text) + except yaml.YAMLError: + return original_content, None + + if parsed is None: + parsed = {} + if not isinstance(parsed, dict): + return original_content, None + + body_content = "".join(lines[closing_index + 1 :]) + return body_content, parsed + + +async def read_note_json_by_external_id( + *, + knowledge_client: KnowledgeEntityReader, + resource_client: NoteResourceReader, + entity_external_id: str, + include_frontmatter: bool = False, +) -> ReadNoteJsonPayload: + """Read and shape one note by exact external ID without identifier resolution. + + The entity response carries the accepted Markdown and its routing metadata. Legacy or + non-note entities may not carry content, so only that explicit ``None`` state falls back to + the raw resource route. Empty accepted Markdown remains a valid response and never triggers + a speculative resource read. + """ + with logfire.span( + "mcp.read_note.shape_response", + domain="mcp", + action="read_note", + phase="shape_response", + ) as span: + entity = await knowledge_client.get_entity(entity_external_id) + content_text = entity.content + resource_fallback = content_text is None + if resource_fallback: + response = await resource_client.read(entity_external_id) + content_text = response.text + + span.set_attribute("read_note.resource_fallback", resource_fallback) + body_content, parsed_frontmatter = parse_opening_frontmatter(content_text) + return { + "title": entity.title, + "permalink": entity.permalink, + "file_path": entity.file_path, + "content": content_text if include_frontmatter else body_content, + "frontmatter": parsed_frontmatter, + } diff --git a/src/basic_memory/mcp/tools/read_note.py b/src/basic_memory/mcp/tools/read_note.py index 42aaaab14..630d7bf46 100644 --- a/src/basic_memory/mcp/tools/read_note.py +++ b/src/basic_memory/mcp/tools/read_note.py @@ -2,9 +2,9 @@ from textwrap import dedent from typing import Any, Annotated, Optional, Literal, cast +from uuid import UUID import logfire -import yaml from loguru import logger from fastmcp import Context @@ -16,6 +16,10 @@ get_project_client, resolve_project_and_path, ) +from basic_memory.mcp.note_reads import ( + parse_opening_frontmatter, + read_note_json_by_external_id, +) from basic_memory.mcp.server import mcp from basic_memory.mcp.tools.search import search_notes from basic_memory.schemas.memory import memory_url_path @@ -39,38 +43,16 @@ def _is_exact_title_match(identifier: str, title: str) -> bool: def _parse_opening_frontmatter(content: str) -> tuple[str, dict[str, Any] | None]: - """Parse opening YAML frontmatter and return (body, frontmatter). - - Mirrors CLI behavior: only parses a frontmatter block at the very top. - If parsing fails or frontmatter is not a mapping, returns body unchanged and None. - """ - original_content = content - lines = content.splitlines(keepends=True) - if not lines or lines[0].strip() != "---": - return original_content, None - - closing_index = None - for i in range(1, len(lines)): - if lines[i].strip() == "---": - closing_index = i - break + """Retain the existing test/import surface for the shared parser.""" + return parse_opening_frontmatter(content) - if closing_index is None: - return original_content, None - fm_text = "".join(lines[1:closing_index]) +def _exact_external_id(identifier: str) -> str | None: + """Return the canonical UUID when the whole identifier is an external ID.""" try: - parsed = yaml.safe_load(fm_text) - except yaml.YAMLError: - return original_content, None - - if parsed is None: - parsed = {} - if not isinstance(parsed, dict): - return original_content, None - - body_content = "".join(lines[closing_index + 1 :]) - return body_content, parsed + return str(UUID(identifier.strip())) + except ValueError: + return None @mcp.tool( @@ -259,25 +241,6 @@ async def read_note( knowledge_client = KnowledgeClient(client, active_project.external_id) resource_client = ResourceClient(client, active_project.external_id) - async def _read_json_payload(entity_id: str) -> dict[str, Any]: - with logfire.span( - "mcp.read_note.shape_response", - domain="mcp", - action="read_note", - phase="shape_response", - ): - entity = await knowledge_client.get_entity(entity_id) - response = await resource_client.read(entity_id) - content_text = response.text - body_content, parsed_frontmatter = _parse_opening_frontmatter(content_text) - return { - "title": entity.title, - "permalink": entity.permalink, - "file_path": entity.file_path, - "content": content_text if include_frontmatter else body_content, - "frontmatter": parsed_frontmatter, - } - def _empty_json_payload() -> dict[str, Any]: return { "title": None, @@ -346,25 +309,52 @@ def _result_file_path(item: dict[str, object]) -> Optional[str]: value = item.get("file_path") return str(value) if value else None - try: - # Try to resolve identifier to entity ID - entity_id = await knowledge_client.resolve_entity(entity_path, strict=True) + def _result_external_id(item: dict[str, object]) -> str | None: + value = item.get("external_id") + return value if isinstance(value, str) and value else None - # Fetch content using entity ID - response = await resource_client.read(entity_id) + if output_format == "json": + exact_external_id = _exact_external_id(entity_path) + if exact_external_id is not None: + return dict( + await read_note_json_by_external_id( + knowledge_client=knowledge_client, + resource_client=resource_client, + entity_external_id=exact_external_id, + include_frontmatter=include_frontmatter, + ) + ) - # If successful, return the content - if response.status_code == 200: + try: + entity_id = await knowledge_client.resolve_entity(entity_path, strict=True) + except Exception as error: # pragma: no cover + logger.info(f"Direct lookup failed for '{entity_path}': {error}") + else: logger.info( - "Returning read_note result from resource: {path}", + "Returning JSON read_note result from entity: {path}", path=entity_path, ) - if output_format == "json": - return await _read_json_payload(entity_id) - return response.text - except Exception as e: # pragma: no cover - logger.info(f"Direct lookup failed for '{entity_path}': {e}") - # Continue to fallback methods + return dict( + await read_note_json_by_external_id( + knowledge_client=knowledge_client, + resource_client=resource_client, + entity_external_id=entity_id, + include_frontmatter=include_frontmatter, + ) + ) + else: + # Text mode intentionally retains the resolve -> resource behavior. + try: + entity_id = await knowledge_client.resolve_entity(entity_path, strict=True) + response = await resource_client.read(entity_id) + if response.status_code == 200: + logger.info( + "Returning read_note result from resource: {path}", + path=entity_path, + ) + return response.text + except Exception as error: # pragma: no cover + logger.info(f"Direct lookup failed for '{entity_path}': {error}") # Fallback 1: Try title search via API, walking fixed-size pages of # title results until an exact match is found or results run out. @@ -405,26 +395,45 @@ def _result_file_path(item: dict[str, object]) -> Optional[str]: logger.info(f"No exact title match found for: {identifier}") break - if result is not None and _result_permalink(result): + if result is not None and output_format == "json": + try: + entity_id = _result_external_id(result) + if entity_id is None and _result_permalink(result) is not None: + entity_id = await knowledge_client.resolve_entity( + _result_permalink(result) or "", strict=True + ) + if entity_id is not None: + logger.info( + f"Found note by exact title search: {_result_permalink(result)}" + ) + return dict( + await read_note_json_by_external_id( + knowledge_client=knowledge_client, + resource_client=resource_client, + entity_external_id=entity_id, + include_frontmatter=include_frontmatter, + ) + ) + except Exception as error: # pragma: no cover + logger.info( + "Failed to fetch content for found title match " + f"{_result_permalink(result)}: {error}" + ) + elif result is not None and _result_permalink(result): try: - # Resolve the permalink to entity ID entity_id = await knowledge_client.resolve_entity( _result_permalink(result) or "", strict=True ) - - # Fetch content using the entity ID response = await resource_client.read(entity_id) - if response.status_code == 200: logger.info( f"Found note by exact title search: {_result_permalink(result)}" ) - if output_format == "json": - return await _read_json_payload(entity_id) return response.text - except Exception as e: # pragma: no cover + except Exception as error: # pragma: no cover logger.info( - f"Failed to fetch content for found title match {_result_permalink(result)}: {e}" + "Failed to fetch content for found title match " + f"{_result_permalink(result)}: {error}" ) # Fallback 2: Text search as a last resort diff --git a/src/basic_memory/read_cache/contract.py b/src/basic_memory/read_cache/contract.py index ecc1f318f..be8b72602 100644 --- a/src/basic_memory/read_cache/contract.py +++ b/src/basic_memory/read_cache/contract.py @@ -73,10 +73,13 @@ class ReadCacheLookup: generation: str payload: bytes | None = None + remaining_ttl_seconds: float | None = None def __post_init__(self) -> None: if not self.generation: raise ValueError("read-cache lookup generation must not be empty") + if self.remaining_ttl_seconds is not None and self.remaining_ttl_seconds < 0: + raise ValueError("read-cache remaining_ttl_seconds must not be negative") @property def is_hit(self) -> bool: diff --git a/src/basic_memory/read_cache/invalidation.py b/src/basic_memory/read_cache/invalidation.py index 0bd445fe2..23d8eb9df 100644 --- a/src/basic_memory/read_cache/invalidation.py +++ b/src/basic_memory/read_cache/invalidation.py @@ -35,7 +35,7 @@ async def invalidate_project_read_cache( except ReadCacheUnavailable as error: # Trigger: an authoritative mutation committed while Redis was unavailable. # Why: failing the request cannot roll the mutation back and would invite - # duplicate retries; the 60-second TTL already bounds stale exposure. + # duplicate retries; the configured response TTL bounds stale exposure. # Outcome: surface prominent telemetry and let the committed write succeed. status = ReadCacheInvalidationStatus.unavailable logger.error( diff --git a/src/basic_memory/read_cache/policy.py b/src/basic_memory/read_cache/policy.py index 133c208e3..7ad1ac73c 100644 --- a/src/basic_memory/read_cache/policy.py +++ b/src/basic_memory/read_cache/policy.py @@ -1,6 +1,6 @@ """Initial semantic read-cache policy.""" -READ_CACHE_TTL_SECONDS = 60 +READ_CACHE_TTL_SECONDS = 300 READ_CACHE_MAX_PAYLOAD_BYTES = 1024 * 1024 DIRECTORY_READ_CACHE_MAX_PAYLOAD_BYTES = 2 * 1024 * 1024 SEARCH_READ_CACHE_TTL_SECONDS = 30 diff --git a/src/basic_memory/read_cache/read_through.py b/src/basic_memory/read_cache/read_through.py index 719ae24d5..5bd81413d 100644 --- a/src/basic_memory/read_cache/read_through.py +++ b/src/basic_memory/read_cache/read_through.py @@ -3,12 +3,14 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from dataclasses import dataclass +from hashlib import sha256 import logfire -from pydantic import BaseModel +from pydantic import BaseModel, ValidationError from basic_memory.read_cache.contract import ( ReadCache, + ReadCacheDataError, ReadCacheInvalidationStatus, ReadCacheKey, ReadCacheUnavailable, @@ -25,6 +27,11 @@ def _record_event(key: ReadCacheKey, event: str) -> None: ) +def _generation_digest(generation: str) -> str: + """Hash the opaque generation token before attaching it to diagnostics.""" + return sha256(generation.encode("utf-8")).hexdigest() + + @dataclass(slots=True) class ReadCacheScope[ModelT: BaseModel]: """Mutable state exchanged with one configured read-cache scope.""" @@ -69,6 +76,14 @@ async def read( "read_cache.read_through", operation=key.operation.value, ) as span: + span.set_attributes( + { + "cache.operation": key.operation.value, + "cache.configured_ttl_seconds": self.ttl_seconds, + "cache.lookup.outcome": "pending", + "cache.store.outcome": "not_attempted", + } + ) try: lookup = await self.backend.lookup(key) except ReadCacheUnavailable: @@ -76,33 +91,56 @@ async def read( # Why: the database or storage path remains authoritative. # Outcome: return fresh data without attempting another cache operation. _record_event(key, "bypass") - span.set_attribute("cache.outcome", "bypass") + span.set_attribute("cache.lookup.outcome", "unavailable") result = ReadCacheScope[ModelT]() yield result result.require_value() return - + except ReadCacheDataError: + # Trigger: Redis returned a structurally invalid cache value. + # Why: corruption must remain fail-fast, but the span still needs a bounded + # terminal outcome instead of retaining its initialization sentinel. + # Outcome: report corruption and preserve the original exception for the caller. + _record_event(key, "corrupt") + span.set_attribute("cache.lookup.outcome", "corrupt") + raise + + lookup_attributes: dict[str, str | int | float] = { + "cache.lookup.outcome": "hit" if lookup.is_hit else "miss", + "cache.generation_digest": _generation_digest(lookup.generation), + } if lookup.payload is not None: + lookup_attributes["cache.payload_bytes"] = len(lookup.payload) + if lookup.remaining_ttl_seconds is not None: + lookup_attributes["cache.remaining_ttl_seconds"] = lookup.remaining_ttl_seconds + try: + cached_value = self.model_type.model_validate_json(lookup.payload) + except ValidationError: + # Trigger: the cache envelope is valid but its typed response payload is not. + # Why: treating invalid data as a hit hides corruption and leaves misleading + # telemetry, while falling back would weaken the fail-fast contract. + # Outcome: mark the lookup corrupt and re-raise the validation error unchanged. + lookup_attributes["cache.lookup.outcome"] = "corrupt" + _record_event(key, "corrupt") + span.set_attributes(lookup_attributes) + raise + _record_event(key, "hit") - span.set_attributes( - { - "cache.outcome": "hit", - "cache.payload_bytes": len(lookup.payload), - } - ) + span.set_attributes(lookup_attributes) yield ReadCacheScope( - value=self.model_type.model_validate_json(lookup.payload), + value=cached_value, cacheable=False, ) return _record_event(key, "miss") + span.set_attributes(lookup_attributes) result = ReadCacheScope[ModelT]() yield result value = result.require_value() if not result.cacheable: _record_event(key, "ineligible") - span.set_attribute("cache.outcome", "ineligible") + span.set_attribute("cache.store.outcome", "ineligible") return payload = value.model_dump_json().encode("utf-8") @@ -110,7 +148,7 @@ async def read( _record_event(key, "oversize") span.set_attributes( { - "cache.outcome": "oversize", + "cache.store.outcome": "oversize", "cache.payload_bytes": len(payload), } ) @@ -125,13 +163,26 @@ async def read( ) except ReadCacheUnavailable: _record_event(key, "store_unavailable") - span.set_attribute("cache.outcome", "store_unavailable") + span.set_attribute("cache.store.outcome", "unavailable") return + except ReadCacheDataError: + # Trigger: Redis returned an invalid result from its guarded store script. + # Why: a Lua contract violation is corruption, not an availability failure; + # swallowing it would conceal a broken cache implementation contract. + # Outcome: preserve the miss, terminate the store outcome, and fail fast. + _record_event(key, "store_corrupt") + span.set_attributes( + { + "cache.store.outcome": "corrupt", + "cache.payload_bytes": len(payload), + } + ) + raise _record_event(key, store_status.value) span.set_attributes( { - "cache.outcome": store_status.value, + "cache.store.outcome": store_status.value, "cache.payload_bytes": len(payload), } ) diff --git a/src/basic_memory/read_cache/redis.py b/src/basic_memory/read_cache/redis.py index b40dd6ca4..fefa445f2 100644 --- a/src/basic_memory/read_cache/redis.py +++ b/src/basic_memory/read_cache/redis.py @@ -46,7 +46,8 @@ else redis.call("SET", KEYS[1], ARGV[1], "EX", ARGV[2]) end -return redis.call("MGET", KEYS[1], KEYS[2]) +local values = redis.call("MGET", KEYS[1], KEYS[2]) +return {values[1], values[2], redis.call("PTTL", KEYS[2])} """ _STORE_IF_CURRENT_SCRIPT = """ if redis.call("GET", KEYS[1]) ~= ARGV[1] then @@ -105,6 +106,16 @@ def _store_status(value: object) -> ReadCacheStoreStatus: raise ReadCacheDataError("Redis returned an invalid cache store result") +def _remaining_ttl_seconds(value: object) -> float: + if isinstance(value, bool) or not isinstance(value, int): + raise ReadCacheDataError("Redis returned an invalid cached payload TTL") + if value == -1: + raise ReadCacheDataError("Redis cached payload has no expiration") + if value < 0: + raise ReadCacheDataError("Redis returned an invalid cached payload TTL") + return value / 1_000 + + class RedisReadCache: """Namespace-bound Redis cache with race-safe generation invalidation. @@ -141,7 +152,7 @@ def _keys(self, key: ReadCacheKey) -> RedisReadCacheKeys: async def lookup(self, key: ReadCacheKey) -> ReadCacheLookup: keys = self._keys(key) try: - generation_value, cached_value = await self._client.eval( + generation_value, cached_value, remaining_ttl_ms = await self._client.eval( _LOOKUP_SCRIPT, 2, keys.generation_key, @@ -162,7 +173,11 @@ async def lookup(self, key: ReadCacheKey) -> ReadCacheLookup: raise ReadCacheDataError("Redis cached payload has an invalid generation envelope") if cached_generation != generation: return ReadCacheLookup(generation=generation_text) - return ReadCacheLookup(generation=generation_text, payload=payload) + return ReadCacheLookup( + generation=generation_text, + payload=payload, + remaining_ttl_seconds=_remaining_ttl_seconds(remaining_ttl_ms), + ) async def store( self, diff --git a/test-int/read_cache/test_redis_read_cache.py b/test-int/read_cache/test_redis_read_cache.py index 0c3e6758c..b0c7f3e23 100644 --- a/test-int/read_cache/test_redis_read_cache.py +++ b/test-int/read_cache/test_redis_read_cache.py @@ -31,6 +31,7 @@ ) from basic_memory.read_cache.redis import ( RedisReadCache, + _remaining_ttl_seconds, _required_bytes, _store_status, create_redis_read_cache_client, @@ -106,11 +107,14 @@ async def test_round_trip_and_ttl_expiry(redis_cache: RedisCacheHarness) -> None assert hit.is_hit assert hit.payload == b"cached entity" assert hit.generation == miss.generation + assert hit.remaining_ttl_seconds is not None + assert 0 < hit.remaining_ttl_seconds <= 1 await asyncio.sleep(1.1) expired = await redis_cache.cache.lookup(key) assert not expired.is_hit assert expired.generation == miss.generation + assert expired.remaining_ttl_seconds is None @pytest.mark.asyncio @@ -414,13 +418,20 @@ async def test_invalidation_leaves_unrelated_redis_data_untouched( @pytest.mark.asyncio async def test_corrupt_redis_values_fail_fast(redis_cache: RedisCacheHarness) -> None: key = _key() - await redis_cache.cache.lookup(key) + lookup = await redis_cache.cache.lookup(key) redis_keys = redis_read_cache_keys( prefix=redis_cache.prefix, namespace=redis_cache.namespace, key=key, ) + await redis_cache.client.set( + redis_keys.data_key, + lookup.generation.encode("ascii") + b"\npersistent payload", + ) + with pytest.raises(ReadCacheDataError, match="has no expiration"): + await redis_cache.cache.lookup(key) + await redis_cache.client.set(redis_keys.data_key, b"missing envelope") with pytest.raises(ReadCacheDataError, match="invalid generation envelope"): await redis_cache.cache.lookup(key) @@ -824,6 +835,8 @@ def test_key_validation_and_canonicalization() -> None: ) with pytest.raises(ValueError, match="generation"): ReadCacheLookup(generation="", payload=b"orphaned") + with pytest.raises(ValueError, match="remaining_ttl_seconds"): + ReadCacheLookup(generation="0" * 32, remaining_ttl_seconds=-0.1) with pytest.raises(ValueError, match="prefix"): redis_read_cache_generation_key( prefix="", @@ -881,6 +894,12 @@ async def test_invalid_store_inputs_fail_before_redis( def test_required_bytes_rejects_non_string_values() -> None: + with pytest.raises(ReadCacheDataError, match="has no expiration"): + _remaining_ttl_seconds(-1) + with pytest.raises(ReadCacheDataError, match="invalid cached payload TTL"): + _remaining_ttl_seconds(-2) + with pytest.raises(ReadCacheDataError, match="invalid cached payload TTL"): + _remaining_ttl_seconds("1000") with pytest.raises(ReadCacheDataError, match="invalid test value"): _required_bytes(1, field="test") with pytest.raises(ReadCacheDataError, match="invalid cache store result"): diff --git a/tests/mcp/test_read_note_request_counts.py b/tests/mcp/test_read_note_request_counts.py new file mode 100644 index 000000000..574395b57 --- /dev/null +++ b/tests/mcp/test_read_note_request_counts.py @@ -0,0 +1,350 @@ +"""Request-count contracts for semantic ``read_note`` JSON paths.""" + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from datetime import UTC, datetime +from types import SimpleNamespace + +import pytest +from httpx import Response + +from basic_memory.mcp.note_reads import read_note_json_by_external_id +from basic_memory.schemas.v2 import EntityResponseV2 + +PROJECT_ID = "11111111-1111-4111-8111-111111111111" +ENTITY_ID = "22222222-2222-4222-8222-222222222222" + + +def _entity( + *, content: str | None = "---\ntitle: Request Count\nstatus: ready\n---\nBody\n" +) -> EntityResponseV2: + now = datetime(2026, 8, 13, tzinfo=UTC) + return EntityResponseV2( + external_id=ENTITY_ID, + id=1, + title="Request Count", + note_type="note", + permalink="notes/request-count", + file_path="notes/Request Count.md", + content=content, + entity_metadata={"title": "Indexed Title", "status": "indexed"}, + created_at=now, + updated_at=now, + ) + + +def _patch_project_routing(monkeypatch: pytest.MonkeyPatch, read_note_module: object) -> None: + @asynccontextmanager + async def fake_get_project_client( + project: str | None, + *, + context: object | None, + project_id: str | None, + ) -> AsyncIterator[tuple[object, SimpleNamespace]]: + del project, context, project_id + yield object(), SimpleNamespace(name="main", external_id=PROJECT_ID, home="/tmp") + + async def fake_resolve_project_and_path( + client: object, + identifier: str, + project: str, + context: object | None, + ) -> tuple[None, str, None]: + del client, project, context + return None, identifier, None + + monkeypatch.setattr(read_note_module, "get_project_client", fake_get_project_client) + monkeypatch.setattr(read_note_module, "resolve_project_and_path", fake_resolve_project_and_path) + monkeypatch.setattr(read_note_module, "validate_project_path", lambda *_args: True) + + +@pytest.mark.asyncio +async def test_exact_uuid_json_reads_entity_once_without_resolve_or_resource( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import importlib + + read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note") + clients_module = importlib.import_module("basic_memory.mcp.clients") + _patch_project_routing(monkeypatch, read_note_module) + calls = {"resolve": 0, "entity": 0, "resource": 0} + + class RecordingKnowledgeClient: + def __init__(self, client: object, project_id: str) -> None: + del client + assert project_id == PROJECT_ID + + async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str: + del identifier, strict + calls["resolve"] += 1 + raise AssertionError("an exact external ID must not be resolved") + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + calls["entity"] += 1 + assert entity_id == ENTITY_ID + return _entity() + + class RecordingResourceClient: + def __init__(self, client: object, project_id: str) -> None: + del client + assert project_id == PROJECT_ID + + async def read(self, entity_id: str) -> Response: + del entity_id + calls["resource"] += 1 + raise AssertionError("accepted entity content must avoid the resource route") + + monkeypatch.setattr(clients_module, "KnowledgeClient", RecordingKnowledgeClient) + monkeypatch.setattr(clients_module, "ResourceClient", RecordingResourceClient) + + result = await read_note_module.read_note(ENTITY_ID, project="main", output_format="json") + + assert isinstance(result, dict) + assert result["content"].strip() == "Body" + assert result["frontmatter"] == {"title": "Request Count", "status": "ready"} + assert calls == {"resolve": 0, "entity": 1, "resource": 0} + + +@pytest.mark.asyncio +async def test_permalink_json_resolves_once_then_reads_entity_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import importlib + + read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note") + clients_module = importlib.import_module("basic_memory.mcp.clients") + _patch_project_routing(monkeypatch, read_note_module) + calls = {"resolve": 0, "entity": 0, "resource": 0} + + class RecordingKnowledgeClient: + def __init__(self, client: object, project_id: str) -> None: + del client, project_id + + async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str: + calls["resolve"] += 1 + assert identifier == "notes/request-count" + assert strict is True + return ENTITY_ID + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + calls["entity"] += 1 + assert entity_id == ENTITY_ID + return _entity() + + class RecordingResourceClient: + def __init__(self, client: object, project_id: str) -> None: + del client, project_id + + async def read(self, entity_id: str) -> Response: + del entity_id + calls["resource"] += 1 + raise AssertionError("accepted entity content must avoid the resource route") + + monkeypatch.setattr(clients_module, "KnowledgeClient", RecordingKnowledgeClient) + monkeypatch.setattr(clients_module, "ResourceClient", RecordingResourceClient) + + result = await read_note_module.read_note( + "notes/request-count", + project="main", + output_format="json", + ) + + assert isinstance(result, dict) + assert result["content"].strip() == "Body" + assert calls == {"resolve": 1, "entity": 1, "resource": 0} + + +@pytest.mark.asyncio +async def test_exact_title_json_uses_search_result_external_id_without_second_resolve( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import importlib + + read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note") + clients_module = importlib.import_module("basic_memory.mcp.clients") + _patch_project_routing(monkeypatch, read_note_module) + calls = {"resolve": 0, "entity": 0, "resource": 0} + + class RecordingKnowledgeClient: + def __init__(self, client: object, project_id: str) -> None: + del client, project_id + + async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str: + calls["resolve"] += 1 + assert identifier == "Request Count" + assert strict is True + raise RuntimeError("force exact-title fallback") + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + calls["entity"] += 1 + assert entity_id == ENTITY_ID + return _entity() + + class RecordingResourceClient: + def __init__(self, client: object, project_id: str) -> None: + del client, project_id + + async def read(self, entity_id: str) -> Response: + del entity_id + calls["resource"] += 1 + raise AssertionError("accepted entity content must avoid the resource route") + + async def fake_search_notes(*, search_type: str, **_kwargs: object) -> dict[str, object]: + assert search_type == "title" + return { + "results": [ + { + "title": "Request Count", + "external_id": ENTITY_ID, + "permalink": "notes/request-count", + "file_path": "notes/Request Count.md", + } + ], + "has_more": False, + } + + monkeypatch.setattr(clients_module, "KnowledgeClient", RecordingKnowledgeClient) + monkeypatch.setattr(clients_module, "ResourceClient", RecordingResourceClient) + monkeypatch.setattr(read_note_module, "search_notes", fake_search_notes) + + result = await read_note_module.read_note( + "Request Count", + project="main", + output_format="json", + ) + + assert isinstance(result, dict) + assert result["content"].strip() == "Body" + assert calls == {"resolve": 1, "entity": 1, "resource": 0} + + +@pytest.mark.asyncio +async def test_text_mode_keeps_resolve_then_resource_behavior( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import importlib + + read_note_module = importlib.import_module("basic_memory.mcp.tools.read_note") + clients_module = importlib.import_module("basic_memory.mcp.clients") + _patch_project_routing(monkeypatch, read_note_module) + calls = {"resolve": 0, "entity": 0, "resource": 0} + + class RecordingKnowledgeClient: + def __init__(self, client: object, project_id: str) -> None: + del client, project_id + + async def resolve_entity(self, identifier: str, *, strict: bool = False) -> str: + calls["resolve"] += 1 + assert identifier == "notes/request-count" + assert strict is True + return ENTITY_ID + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + del entity_id + calls["entity"] += 1 + raise AssertionError("text mode must not load the entity response") + + class RecordingResourceClient: + def __init__(self, client: object, project_id: str) -> None: + del client, project_id + + async def read(self, entity_id: str) -> Response: + calls["resource"] += 1 + assert entity_id == ENTITY_ID + return Response(200, text="raw text-mode Markdown") + + monkeypatch.setattr(clients_module, "KnowledgeClient", RecordingKnowledgeClient) + monkeypatch.setattr(clients_module, "ResourceClient", RecordingResourceClient) + + result = await read_note_module.read_note("notes/request-count", project="main") + + assert result == "raw text-mode Markdown" + assert calls == {"resolve": 1, "entity": 0, "resource": 1} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("accepted_content", ["accepted content", ""]) +async def test_exact_id_helper_does_not_read_resource_for_present_content( + accepted_content: str, +) -> None: + class EntityReader: + calls = 0 + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + self.calls += 1 + assert entity_id == ENTITY_ID + return _entity(content=accepted_content) + + class ResourceReader: + calls = 0 + + async def read(self, entity_id: str) -> Response: + del entity_id + self.calls += 1 + raise AssertionError("present content must not fall back to resource") + + entity_reader = EntityReader() + resource_reader = ResourceReader() + result = await read_note_json_by_external_id( + knowledge_client=entity_reader, + resource_client=resource_reader, + entity_external_id=ENTITY_ID, + ) + + assert result["content"] == accepted_content + assert entity_reader.calls == 1 + assert resource_reader.calls == 0 + + +@pytest.mark.asyncio +async def test_exact_id_helper_reads_resource_once_only_when_content_is_absent() -> None: + class EntityReader: + calls = 0 + + async def get_entity(self, entity_id: str) -> EntityResponseV2: + self.calls += 1 + assert entity_id == ENTITY_ID + return _entity(content=None) + + class ResourceReader: + calls = 0 + + async def read(self, entity_id: str) -> Response: + self.calls += 1 + assert entity_id == ENTITY_ID + return Response(200, text="---\nlegacy: true\n---\nlegacy body\n") + + entity_reader = EntityReader() + resource_reader = ResourceReader() + result = await read_note_json_by_external_id( + knowledge_client=entity_reader, + resource_client=resource_reader, + entity_external_id=ENTITY_ID, + ) + + assert result["content"].strip() == "legacy body" + assert result["frontmatter"] == {"legacy": True} + assert entity_reader.calls == 1 + assert resource_reader.calls == 1 + + +@pytest.mark.asyncio +async def test_exact_id_helper_does_not_fabricate_frontmatter_from_entity_metadata() -> None: + class EntityReader: + async def get_entity(self, entity_id: str) -> EntityResponseV2: + assert entity_id == ENTITY_ID + return _entity(content="plain body\n") + + class ResourceReader: + async def read(self, entity_id: str) -> Response: + del entity_id + raise AssertionError("present content must not fall back to resource") + + result = await read_note_json_by_external_id( + knowledge_client=EntityReader(), + resource_client=ResourceReader(), + entity_external_id=ENTITY_ID, + ) + + assert result["content"] == "plain body\n" + assert result["frontmatter"] is None diff --git a/tests/mcp/test_tool_read_note.py b/tests/mcp/test_tool_read_note.py index 610f1912a..fa8c29ae4 100644 --- a/tests/mcp/test_tool_read_note.py +++ b/tests/mcp/test_tool_read_note.py @@ -565,6 +565,8 @@ async def get_entity(self, entity_id: str): title="TODO", permalink="personal/main/todo", file_path="TODO.md", + content="---\ntitle: TODO\n---\n\n# TODO - Priorities & Tasks\n", + entity_metadata={"title": "TODO"}, ) class FakeResourceClient: @@ -572,11 +574,7 @@ def __init__(self, client, project_id): assert project_id == expected_uuid async def read(self, entity_id: str): - assert entity_id == "entity-1" - return SimpleNamespace( - status_code=200, - text="---\ntitle: TODO\n---\n\n# TODO - Priorities & Tasks\n", - ) + raise AssertionError(f"accepted content must avoid resource read for {entity_id}") monkeypatch.setattr( project_context, diff --git a/tests/mcp/test_tool_telemetry.py b/tests/mcp/test_tool_telemetry.py index 36bcc96a2..2d8cf83d7 100644 --- a/tests/mcp/test_tool_telemetry.py +++ b/tests/mcp/test_tool_telemetry.py @@ -120,8 +120,8 @@ async def test_read_note_emits_root_operation_and_project_context( ) span_names = [name for name, _ in spans] assert "api.request.knowledge.resolve_entity" in span_names - assert "api.request.resource.get_content" in span_names assert "api.request.knowledge.get_entity" in span_names + assert "api.request.resource.get_content" not in span_names assert _contains_span_attrs( spans, "routing.client_session", diff --git a/tests/test_read_cache_telemetry.py b/tests/test_read_cache_telemetry.py new file mode 100644 index 000000000..f87c2450f --- /dev/null +++ b/tests/test_read_cache_telemetry.py @@ -0,0 +1,323 @@ +"""Telemetry contracts for typed semantic read-through caching.""" + +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from hashlib import sha256 + +import pytest +from pydantic import BaseModel, ValidationError + +from basic_memory.read_cache import ( + ModelReadCache, + ReadCacheDataError, + ReadCacheInvalidationStatus, + ReadCacheKey, + ReadCacheLookup, + ReadCacheOperation, + ReadCacheStoreStatus, + ReadCacheUnavailable, + read_cache_request_digest, +) +from basic_memory.read_cache import read_through +from basic_memory.read_cache.policy import ( + READ_CACHE_TTL_SECONDS, + SEARCH_READ_CACHE_TTL_SECONDS, +) + +PROJECT_ID = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +GENERATION = "1" * 32 + + +class CachedValue(BaseModel): + title: str + + +def test_production_ttls_keep_search_shorter_than_ordinary_reads() -> None: + assert READ_CACHE_TTL_SECONDS == 300 + assert SEARCH_READ_CACHE_TTL_SECONDS == 30 + + +@dataclass(slots=True) +class RecordedSpan: + name: str + attributes: dict[str, object] = field(default_factory=dict) + + def set_attribute(self, name: str, value: object) -> None: + self.attributes[name] = value + + def set_attributes(self, attributes: dict[str, object]) -> None: + self.attributes.update(attributes) + + +@dataclass(slots=True) +class RecordingCache: + lookup_result: ReadCacheLookup | None = None + lookup_error: ReadCacheUnavailable | ReadCacheDataError | None = None + store_status: ReadCacheStoreStatus = ReadCacheStoreStatus.stored + store_error: ReadCacheUnavailable | ReadCacheDataError | None = None + store_ttls: list[int] = field(default_factory=list) + + async def lookup(self, key: ReadCacheKey) -> ReadCacheLookup: + del key + if self.lookup_error is not None: + raise self.lookup_error + if self.lookup_result is None: + raise AssertionError("test cache requires a lookup result or error") + return self.lookup_result + + async def store( + self, + key: ReadCacheKey, + lookup: ReadCacheLookup, + payload: bytes, + *, + ttl_seconds: int, + ) -> ReadCacheStoreStatus: + del key, lookup, payload + self.store_ttls.append(ttl_seconds) + if self.store_error is not None: + raise self.store_error + return self.store_status + + async def invalidate_project(self, project_id: str) -> ReadCacheInvalidationStatus: + del project_id + return ReadCacheInvalidationStatus.invalidated + + +def _key() -> ReadCacheKey: + return ReadCacheKey( + project_id=PROJECT_ID, + operation=ReadCacheOperation.entity, + request_digest=read_cache_request_digest("entity-1"), + ) + + +def _capture_telemetry( + monkeypatch: pytest.MonkeyPatch, +) -> tuple[list[RecordedSpan], list[tuple[str, dict[str, str]]]]: + spans: list[RecordedSpan] = [] + events: list[tuple[str, dict[str, str]]] = [] + + @contextmanager + def fake_span(name: str, **attributes: object) -> Iterator[RecordedSpan]: + span = RecordedSpan(name=name, attributes=dict(attributes)) + spans.append(span) + yield span + + class Counter: + def add(self, amount: int, *, attributes: dict[str, str]) -> None: + assert amount == 1 + events.append(("basic_memory_read_cache_events_total", attributes)) + + def fake_metric_counter(name: str) -> Counter: + assert name == "basic_memory_read_cache_events_total" + return Counter() + + monkeypatch.setattr(read_through.logfire, "span", fake_span) + monkeypatch.setattr(read_through.logfire, "metric_counter", fake_metric_counter) + return spans, events + + +@pytest.mark.asyncio +async def test_hit_telemetry_reports_lookup_details_without_store( + monkeypatch: pytest.MonkeyPatch, +) -> None: + spans, events = _capture_telemetry(monkeypatch) + payload = CachedValue(title="cached").model_dump_json().encode("utf-8") + backend = RecordingCache( + lookup_result=ReadCacheLookup( + generation=GENERATION, + payload=payload, + remaining_ttl_seconds=241.25, + ) + ) + cache = ModelReadCache( + backend=backend, + model_type=CachedValue, + ttl_seconds=300, + max_payload_bytes=1_024, + ) + + async with cache.read(key=_key()) as cached: + assert cached.value == CachedValue(title="cached") + + assert len(spans) == 1 + assert spans[0].attributes == { + "operation": "entity", + "cache.operation": "entity", + "cache.configured_ttl_seconds": 300, + "cache.lookup.outcome": "hit", + "cache.store.outcome": "not_attempted", + "cache.generation_digest": sha256(GENERATION.encode("utf-8")).hexdigest(), + "cache.payload_bytes": len(payload), + "cache.remaining_ttl_seconds": 241.25, + } + assert events == [ + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "hit"}) + ] + assert backend.store_ttls == [] + + +@pytest.mark.asyncio +async def test_miss_telemetry_keeps_lookup_and_store_outcomes_separate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + spans, events = _capture_telemetry(monkeypatch) + backend = RecordingCache(lookup_result=ReadCacheLookup(generation=GENERATION)) + cache = ModelReadCache( + backend=backend, + model_type=CachedValue, + ttl_seconds=300, + max_payload_bytes=1_024, + ) + + async with cache.read(key=_key()) as cached: + cached.value = CachedValue(title="authoritative") + + payload = CachedValue(title="authoritative").model_dump_json().encode("utf-8") + assert len(spans) == 1 + assert spans[0].attributes == { + "operation": "entity", + "cache.operation": "entity", + "cache.configured_ttl_seconds": 300, + "cache.lookup.outcome": "miss", + "cache.store.outcome": "stored", + "cache.generation_digest": sha256(GENERATION.encode("utf-8")).hexdigest(), + "cache.payload_bytes": len(payload), + } + assert events == [ + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "miss"}), + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "stored"}), + ] + assert backend.store_ttls == [300] + + +@pytest.mark.asyncio +async def test_unavailable_lookup_remains_fail_open_and_reports_bypass( + monkeypatch: pytest.MonkeyPatch, +) -> None: + spans, events = _capture_telemetry(monkeypatch) + backend = RecordingCache(lookup_error=ReadCacheUnavailable("Redis unavailable")) + cache = ModelReadCache( + backend=backend, + model_type=CachedValue, + ttl_seconds=300, + max_payload_bytes=1_024, + ) + + async with cache.read(key=_key()) as cached: + cached.value = CachedValue(title="authoritative") + + assert spans[0].attributes["cache.lookup.outcome"] == "unavailable" + assert spans[0].attributes["cache.store.outcome"] == "not_attempted" + assert spans[0].attributes["cache.configured_ttl_seconds"] == 300 + assert events == [ + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "bypass"}) + ] + assert backend.store_ttls == [] + + +@pytest.mark.asyncio +async def test_corrupt_lookup_error_is_terminal_and_remains_fail_fast( + monkeypatch: pytest.MonkeyPatch, +) -> None: + spans, events = _capture_telemetry(monkeypatch) + backend = RecordingCache(lookup_error=ReadCacheDataError("invalid Redis envelope")) + cache = ModelReadCache( + backend=backend, + model_type=CachedValue, + ttl_seconds=300, + max_payload_bytes=1_024, + ) + + with pytest.raises(ReadCacheDataError, match="invalid Redis envelope"): + async with cache.read(key=_key()): + raise AssertionError("a corrupt lookup must not yield") + + assert spans[0].attributes["cache.lookup.outcome"] == "corrupt" + assert spans[0].attributes["cache.store.outcome"] == "not_attempted" + assert events == [ + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "corrupt"}) + ] + assert backend.store_ttls == [] + + +@pytest.mark.asyncio +async def test_invalid_cached_model_is_terminal_and_remains_fail_fast( + monkeypatch: pytest.MonkeyPatch, +) -> None: + spans, events = _capture_telemetry(monkeypatch) + payload = b"not-json" + backend = RecordingCache( + lookup_result=ReadCacheLookup( + generation=GENERATION, + payload=payload, + remaining_ttl_seconds=123.5, + ) + ) + cache = ModelReadCache( + backend=backend, + model_type=CachedValue, + ttl_seconds=300, + max_payload_bytes=1_024, + ) + + with pytest.raises(ValidationError): + async with cache.read(key=_key()): + raise AssertionError("an invalid cached model must not yield") + + assert spans[0].attributes == { + "operation": "entity", + "cache.operation": "entity", + "cache.configured_ttl_seconds": 300, + "cache.lookup.outcome": "corrupt", + "cache.store.outcome": "not_attempted", + "cache.generation_digest": sha256(GENERATION.encode("utf-8")).hexdigest(), + "cache.payload_bytes": len(payload), + "cache.remaining_ttl_seconds": 123.5, + } + assert events == [ + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "corrupt"}) + ] + assert backend.store_ttls == [] + + +@pytest.mark.asyncio +async def test_corrupt_store_error_is_terminal_and_remains_fail_fast( + monkeypatch: pytest.MonkeyPatch, +) -> None: + spans, events = _capture_telemetry(monkeypatch) + backend = RecordingCache( + lookup_result=ReadCacheLookup(generation=GENERATION), + store_error=ReadCacheDataError("invalid Redis store result"), + ) + cache = ModelReadCache( + backend=backend, + model_type=CachedValue, + ttl_seconds=300, + max_payload_bytes=1_024, + ) + + with pytest.raises(ReadCacheDataError, match="invalid Redis store result"): + async with cache.read(key=_key()) as cached: + cached.value = CachedValue(title="authoritative") + + payload = CachedValue(title="authoritative").model_dump_json().encode("utf-8") + assert spans[0].attributes == { + "operation": "entity", + "cache.operation": "entity", + "cache.configured_ttl_seconds": 300, + "cache.lookup.outcome": "miss", + "cache.store.outcome": "corrupt", + "cache.generation_digest": sha256(GENERATION.encode("utf-8")).hexdigest(), + "cache.payload_bytes": len(payload), + } + assert events == [ + ("basic_memory_read_cache_events_total", {"operation": "entity", "event": "miss"}), + ( + "basic_memory_read_cache_events_total", + {"operation": "entity", "event": "store_corrupt"}, + ), + ] + assert backend.store_ttls == [300]