Coverage for astrocyte/recall/merge_result.py: 87%

1"""Merge federated / external MemoryHit rows into a RecallResult (engine & tiered paths).""" 

2 

3from __future__ import annotations 

4 

5from astrocyte.types import MemoryHit, RecallResult 

6 

7 

8def merge_external_into_recall_result( 

9 result: RecallResult, 

10 external: list[MemoryHit], 

11 max_results: int, 

12) -> RecallResult: 

13 """Combine external hits with ``result.hits`` by score, dedupe by text, trim to ``max_results``.""" 

14 combined = list(external) + list(result.hits) 

15 combined.sort(key=lambda h: h.score, reverse=True) 

16 seen: set[str] = set() 

17 deduped: list[MemoryHit] = [] 

18 for h in combined: 

19 if h.text in seen: 

20 continue 

21 seen.add(h.text) 

22 deduped.append(h) 

23 if len(deduped) >= max_results: 

24 break 

25 return RecallResult( 

26 hits=deduped, 

27 total_available=result.total_available + len(external), 

28 truncated=result.truncated, 

29 trace=result.trace, 

30 )