Coverage for node / src / stigmem_node / routes / recall / common.py: 96%

1"""Shared recall route helpers and compatibility utilities.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import sys 

7from datetime import UTC, datetime 

8from typing import Any 

9 

10from fastapi import APIRouter 

11 

12from ...auth import Identity 

13from ...models.facts import FactRecord, row_to_record 

14from ..cid_integrity import enforce_read_path_cid 

15 

16logger = logging.getLogger("stigmem.recall") 

17 

18router = APIRouter(prefix="/v1/recall", tags=["recall"]) 

19 

20def _public_module() -> Any: 

21 """Return the public recall module so test monkey-patches stay visible.""" 

22 return sys.modules["stigmem_node.routes.recall"] 

23 

24def _now_iso() -> str: 

25 return datetime.now(UTC).isoformat() 

26 

27 

28def _estimate_tokens(record: FactRecord) -> int: 

29 """Rough token estimate: 4 chars ≈ 1 token (GPT tokeniser heuristic).""" 

30 text = f"{record.entity} {record.relation} {record.value.v or ''} {record.source}" 

31 return max(1, len(text) // 4) 

32 

33 

34def _recency_score(timestamp_str: str) -> float: 

35 """Normalised recency: 1.0 = now, 0.0 = ≥1 year old.""" 

36 try: 

37 ts = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00")) 

38 if ts.tzinfo is None: 

39 ts = ts.replace(tzinfo=UTC) 

40 days_old = (datetime.now(UTC) - ts).days 

41 return max(0.0, 1.0 - days_old / 365.0) 

42 except Exception: 

43 return 0.5 

44 

45 

46def _fetch_facts_by_ids( 

47 conn: Any, 

48 fact_ids: list[str], 

49) -> dict[str, FactRecord]: 

50 """Bulk-fetch facts by ID; returns {id: FactRecord}.""" 

51 if not fact_ids: 

52 return {} 

53 placeholders = ",".join("?" * len(fact_ids)) 

54 rows = conn.execute( 

55 f""" 

56 SELECT f.*, 

57 COALESCE(fvo.valid_until, f.valid_until) AS projected_valid_until, 

58 COALESCE(fvo.confidence, f.confidence) AS projected_confidence, 

59 COALESCE(fgm.garden_id, f.garden_id) AS projected_garden_id, 

60 COALESCE(fqs.quarantine_status, f.quarantine_status) 

61 AS projected_quarantine_status, 

62 COALESCE(fqs.quarantine_garden_id, f.quarantine_garden_id) 

63 AS projected_quarantine_garden_id, 

64 ( 

65 SELECT fca.cid 

66 FROM fact_cid_aliases fca 

67 WHERE fca.fact_id = f.id 

68 ORDER BY fca.cid 

69 LIMIT 1 

70 ) AS projected_cid 

71 FROM facts f 

72 LEFT JOIN fact_validity_overrides fvo ON fvo.fact_id = f.id 

73 LEFT JOIN fact_garden_membership fgm ON fgm.fact_id = f.id 

74 LEFT JOIN fact_quarantine_status fqs ON fqs.fact_id = f.id 

75 WHERE f.id IN ({placeholders}) 

76 """, # noqa: S608 # nosec B608 

77 fact_ids, 

78 ).fetchall() 

79 for row in rows: 

80 enforce_read_path_cid(row) 

81 return {row["id"]: row_to_record(row) for row in rows} 

82 

83 

84def _write_recall_audit( 

85 conn: Any, 

86 recall_id: str, 

87 identity: Identity, 

88 query_hash: str, 

89 scope: str, 

90 token_budget: int, 

91 facts_returned: int, 

92 tokens_used: int, 

93 truncated: bool, 

94) -> None: 

95 now = _now_iso() 

96 try: 

97 conn.execute( 

98 """ 

99 INSERT INTO recall_audit_log 

100 (id, entity_uri, query_hash, scope, token_budget, 

101 facts_returned, tokens_used, truncated, tenant_id, ts) 

102 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 

103 """, 

104 ( 

105 recall_id, 

106 identity.entity_uri, 

107 query_hash, 

108 scope, 

109 token_budget, 

110 facts_returned, 

111 tokens_used, 

112 1 if truncated else 0, 

113 identity.tenant_id, 

114 now, 

115 ), 

116 ) 

117 except Exception as exc: 

118 logger.error("Failed to write recall_audit_log: %s", exc)