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

1"""BFS graph traversal — neighbors() implementation, spec §20.2. 

2 

3Exported surface: 

4 bfs_neighbors(conn, seed_entity, max_depth, scope, tenant_id, ...) -> list[NeighborEntry] 

5 

6Invariants honored: 

7 - Only traverses edges with confidence >= min_confidence and valid_until not expired. 

8 - Filters garden-tagged edges against caller ACL (§17.3). 

9 - Blocks federated edges (received_from is not null) if caller lacks federate perm (§19.5.2). 

10 - Relation filter supports prefix-glob only (e.g. "memory:*"), not full regex. 

11 - De-duplicates: each neighbor entity appears once at its shortest hop distance. 

12""" 

13 

14from __future__ import annotations 

15 

16from dataclasses import dataclass, field 

17from datetime import UTC, datetime 

18from typing import TYPE_CHECKING, Any 

19 

20if TYPE_CHECKING: 

21 from ..auth import Identity 

22 

23MAX_DEPTH = 3 

24 

25 

26@dataclass 

27class NeighborEntry: 

28 entity: str 

29 relation: str 

30 hops: int 

31 confidence: float 

32 source_trust: float | None 

33 path: list[str] = field(default_factory=list) 

34 

35 

36def bfs_neighbors( 

37 conn: Any, 

38 seed_entity: str, 

39 max_depth: int, 

40 scope: str, 

41 tenant_id: str, 

42 relation_filter: str | None = None, 

43 min_confidence: float = 0.1, 

44 min_trust: float = 0.0, 

45 identity: Identity | None = None, 

46) -> list[NeighborEntry]: 

47 """BFS over entity_edges, returning all neighbors within max_depth hops. 

48 

49 path in each NeighborEntry is the list of traversed entity URIs from the 

50 seed up to (and including) the seed, excluding the neighbor itself. 

51 Shortest path is reported when an entity is reachable via multiple routes. 

52 """ 

53 now = datetime.now(UTC).isoformat() 

54 

55 results: list[NeighborEntry] = [] 

56 # visited[entity] = path_from_seed_to_entity_not_including_entity 

57 visited: dict[str, list[str]] = {seed_entity: []} 

58 frontier: list[str] = [seed_entity] 

59 

60 for hop in range(1, max_depth + 1): 

61 if not frontier: 

62 break 

63 

64 next_frontier: list[str] = [] 

65 

66 for current_entity in frontier: 

67 path_to_current = visited[current_entity] 

68 # Path stored in neighbor = nodes from seed up to and including current 

69 neighbor_path = [*path_to_current, current_entity] 

70 

71 edges = _fetch_incident_edges( 

72 conn, 

73 current_entity, 

74 scope, 

75 tenant_id, 

76 now, 

77 min_confidence, 

78 min_trust, 

79 relation_filter, 

80 identity, 

81 ) 

82 

83 for edge in edges: 

84 neighbor = edge["object"] if edge["subject"] == current_entity else edge["subject"] 

85 if neighbor in visited: 

86 continue 

87 

88 entry = NeighborEntry( 

89 entity=neighbor, 

90 relation=edge["relation"], 

91 hops=hop, 

92 confidence=edge["confidence"], 

93 source_trust=edge["source_trust"], 

94 path=neighbor_path, 

95 ) 

96 results.append(entry) 

97 visited[neighbor] = neighbor_path 

98 next_frontier.append(neighbor) 

99 

100 frontier = next_frontier 

101 

102 return results 

103 

104 

105def _match_relation(relation: str, pattern: str) -> bool: 

106 """Prefix-glob match — 'memory:*' matches any relation starting with 'memory:'.""" 

107 if pattern.endswith("*"): 

108 return relation.startswith(pattern[:-1]) 

109 return relation == pattern 

110 

111 

112def _fetch_incident_edges( 

113 conn: Any, 

114 entity: str, 

115 scope: str, 

116 tenant_id: str, 

117 now: str, 

118 min_confidence: float, 

119 min_trust: float, 

120 relation_filter: str | None, 

121 identity: Identity | None, 

122) -> list[Any]: 

123 """Fetch and filter edges incident on entity from both traversal directions. 

124 

125 Pre-filters by confidence + trust in SQL for efficiency. Garden ACL, 

126 federation scope, and relation filter are applied in Python after fetch. 

127 """ 

128 params: list[Any] = [entity, entity, scope, tenant_id, min_confidence, now] 

129 sql = """ 

130 SELECT id, subject, relation, object, scope, garden_id, 

131 received_from, confidence, source_trust 

132 FROM entity_edges 

133 WHERE (subject = ? OR object = ?) 

134 AND scope = ? 

135 AND tenant_id = ? 

136 AND confidence >= ? 

137 AND (valid_until IS NULL OR valid_until > ?) 

138 """ 

139 

140 if min_trust > 0.0: 140 ↛ 142line 140 didn't jump to line 142 because the condition on line 140 was never true

141 # Edges with NULL source_trust pass through (trust not yet computed) 

142 sql += " AND (source_trust IS NULL OR source_trust >= ?)" 

143 params.append(min_trust) 

144 

145 rows = conn.execute(sql, params).fetchall() 

146 

147 # Garden ACL (§17.3): resolve the caller's read scope ONCE (batched, audit M3 

148 # secure-path), then filter edges in-memory via the shared predicate. 

149 # Anonymous (identity is None) callers skip garden filtering, as before. 

150 from ..fact_visibility import caller_read_scope 

151 

152 read_scope = caller_read_scope(identity) if identity is not None else None 

153 

154 filtered: list[Any] = [] 

155 for row in rows: 

156 # Garden ACL (§17.3): hide edges in gardens the caller cannot see 

157 garden_id = row["garden_id"] 

158 if ( 

159 read_scope is not None 

160 and garden_id is not None 

161 and not read_scope.garden_allows(garden_id) 

162 ): 

163 continue 

164 

165 # Federation filter (§19.5.2): edges from remote nodes require federate perm 

166 received_from = row["received_from"] 

167 if received_from is not None and (identity is None or not identity.can_federate()): 

168 continue 

169 

170 # Relation prefix-glob filter 

171 if relation_filter is not None and not _match_relation(row["relation"], relation_filter): 

172 continue 

173 

174 filtered.append(row) 

175 

176 return filtered