Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-skills/src/lexigram/ai/skills/caching/skill_cache.py: 38%

1"""SkillResultCache — two-tier in-memory and optional backend result caching.""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6from typing import TYPE_CHECKING, Any 

7 

8from lexigram.contracts.ai.skills import SkillResult 

9from lexigram.logging import ( 

10 get_logger, 

11) 

12from lexigram.serialization.backends.json import dumps_str, loads 

13 

14if TYPE_CHECKING: 

15 from lexigram.contracts.infra.cache import CacheBackendProtocol 

16 

17logger = get_logger(__name__) 

18 

19_SENTINEL = object() 

20 

21 

22def _cache_key(skill_name: str, params: dict[str, Any]) -> str: 

23 """Build a deterministic cache key from skill name and parameters. 

24 

25 Args: 

26 skill_name: The name of the skill. 

27 params: Parameters dict (must be JSON-serialisable). 

28 

29 Returns: 

30 A hex-digest string suitable for use as a dict key or cache key. 

31 """ 

32 serialised = dumps_str({"skill": skill_name, "params": params}, sort_keys=True) 

33 return hashlib.sha256(serialised.encode()).hexdigest() 

34 

35 

36class SkillResultCache: 

37 """Two-tier result cache for skill executions. 

38 

39 The first tier is an in-process ``dict``; the second tier is an optional 

40 :class:`CacheBackendProtocol` (e.g. Redis). Lookups check the in-process dict 

41 first; on a miss they query the backend and populate the in-process dict. 

42 

43 Args: 

44 backend: Optional async cache backend for cross-process caching. 

45 ttl_seconds: Time-to-live for backend entries in seconds. 

46 """ 

47 

48 def __init__( 

49 self, 

50 backend: CacheBackendProtocol | None = None, 

51 ttl_seconds: int = 3600, 

52 ) -> None: 

53 """Initialise the cache. 

54 

55 Args: 

56 backend: Optional CacheBackendProtocol for distributed caching. 

57 ttl_seconds: TTL for backend cache entries. 

58 """ 

59 self._local: dict[str, SkillResult] = {} 

60 self._backend = backend 

61 self._ttl = ttl_seconds 

62 

63 async def get(self, skill_name: str, params: dict[str, Any]) -> SkillResult | None: 

64 """Return a cached result for the given skill invocation. 

65 

66 Args: 

67 skill_name: Name of the skill. 

68 params: Parameters the skill was called with. 

69 

70 Returns: 

71 The cached :class:`SkillResult` or ``None`` when not cached. 

72 """ 

73 key = _cache_key(skill_name, params) 

74 

75 hit = self._local.get(key) 

76 if hit is not None: 

77 logger.debug("skill_cache_local_hit", skill=skill_name, key=key[:8]) 

78 return hit 

79 

80 if self._backend is not None: 

81 raw = await self._backend.get(key) 

82 if raw is not None: 

83 try: 

84 data = loads(raw) # type: ignore[arg-type] 

85 result = SkillResult(**data) 

86 self._local[key] = result 

87 logger.debug( 

88 "skill_cache_backend_hit", skill=skill_name, key=key[:8] 

89 ) 

90 return result 

91 except Exception as exc: # noqa: BLE001 

92 logger.warning( 

93 "skill_cache_deserialise_error", 

94 key=key[:8], 

95 error=str(exc), 

96 ) 

97 

98 return None 

99 

100 async def set( 

101 self, skill_name: str, params: dict[str, Any], result: SkillResult 

102 ) -> None: 

103 """Store a skill result in both cache tiers. 

104 

105 Args: 

106 skill_name: Name of the skill. 

107 params: Parameters the skill was called with. 

108 result: The :class:`SkillResult` to cache. 

109 """ 

110 key = _cache_key(skill_name, params) 

111 self._local[key] = result 

112 

113 if self._backend is not None: 

114 try: 

115 serialised = dumps_str( 

116 { 

117 "skill_name": result.skill_name, 

118 "success": result.success, 

119 "output": result.output, 

120 "error": result.error, 

121 "metadata": result.metadata, 

122 } 

123 ) 

124 await self._backend.set(key, serialised, ttl=self._ttl) 

125 except Exception as exc: # noqa: BLE001 

126 logger.warning("skill_cache_store_error", key=key[:8], error=str(exc)) 

127 

128 def invalidate(self, skill_name: str, params: dict[str, Any]) -> None: 

129 """Remove a single entry from the local in-process cache. 

130 

131 Args: 

132 skill_name: Name of the skill. 

133 params: Parameters identifying the entry to remove. 

134 """ 

135 key = _cache_key(skill_name, params) 

136 self._local.pop(key, None) 

137 

138 def clear(self) -> None: 

139 """Remove all entries from the local in-process cache.""" 

140 self._local.clear()