Coverage for node / src / stigmem_node / lifecycle / decay.py: 95%

1"""Configurable decay sweeper — marks stale facts as expired (Phase 6).""" 

2 

3from __future__ import annotations 

4 

5import uuid 

6from datetime import UTC, datetime, timedelta 

7from typing import Any 

8 

9from ..db import db 

10from ..settings import settings 

11from .immutability import set_fact_validity_override 

12 

13 

14def _resolve_effective_ttl(ttl_seconds: int | None) -> int | None: 

15 """Resolve the effective TTL: explicit arg overrides settings; 0 means "expire all".""" 

16 if ttl_seconds is not None: 

17 return ttl_seconds # 0 is valid: expire everything 

18 if settings.decay_ttl_seconds > 0: 18 ↛ 19line 18 didn't jump to line 19 because the condition on line 18 was never true

19 return settings.decay_ttl_seconds 

20 return None 

21 

22 

23def _resolve_effective_min_conf(min_confidence: float | None) -> float | None: 

24 """Resolve the effective confidence floor: explicit arg overrides settings.""" 

25 if min_confidence is not None and min_confidence > 0.0: 

26 return min_confidence 

27 if settings.decay_min_confidence > 0.0: 27 ↛ 28line 27 didn't jump to line 28 because the condition on line 27 was never true

28 return settings.decay_min_confidence 

29 return None 

30 

31 

32def _select_ttl_candidates( 

33 conn: Any, effective_ttl: int, scope: str | None, now_dt: datetime, tenant_id: str 

34) -> list[str]: 

35 """Return fact ids whose timestamp is older than (now - effective_ttl).""" 

36 cutoff = (now_dt - timedelta(seconds=effective_ttl)).isoformat() 

37 sql = ( 

38 "SELECT f.id FROM facts f " 

39 "LEFT JOIN fact_validity_overrides fvo ON fvo.fact_id = f.id " 

40 "WHERE f.timestamp <= ? " 

41 "AND f.tenant_id = ? " 

42 "AND COALESCE(fvo.valid_until, f.valid_until) IS NULL " 

43 "AND NOT (entity LIKE 'stigmem:%' AND entity NOT LIKE 'stigmem://%') " 

44 "AND NOT (relation LIKE 'stigmem:%' AND relation NOT LIKE 'stigmem://%')" 

45 ) 

46 params: list[Any] = [cutoff, tenant_id] 

47 if scope: 

48 sql += " AND scope = ?" 

49 params.append(scope) 

50 return [r["id"] for r in conn.execute(sql, params).fetchall()] 

51 

52 

53def _select_confidence_candidates( 

54 conn: Any, effective_min_conf: float, scope: str | None, now: str, tenant_id: str 

55) -> list[str]: 

56 """Return active fact ids whose confidence is below the floor.""" 

57 sql = ( 

58 "SELECT f.id FROM facts f " 

59 "LEFT JOIN fact_validity_overrides fvo ON fvo.fact_id = f.id " 

60 "WHERE COALESCE(fvo.confidence, f.confidence) < ? " 

61 "AND f.tenant_id = ? " 

62 "AND COALESCE(fvo.confidence, f.confidence) > 0.0 " 

63 "AND (COALESCE(fvo.valid_until, f.valid_until) IS NULL " 

64 "OR COALESCE(fvo.valid_until, f.valid_until) > ?) " 

65 "AND NOT (entity LIKE 'stigmem:%' AND entity NOT LIKE 'stigmem://%') " 

66 "AND NOT (relation LIKE 'stigmem:%' AND relation NOT LIKE 'stigmem://%')" 

67 ) 

68 params: list[Any] = [effective_min_conf, tenant_id, now] 

69 if scope: 

70 sql += " AND scope = ?" 

71 params.append(scope) 

72 return [r["id"] for r in conn.execute(sql, params).fetchall()] 

73 

74 

75def _apply_decay(conn: Any, candidates: list[str], conf_ids: list[str], now: str) -> None: 

76 """Persist decay: mark candidates expired, log confidence retractions, sync graph.""" 

77 for fact_id in candidates: 

78 set_fact_validity_override( 

79 conn, 

80 fact_id=fact_id, 

81 valid_until=now, 

82 reason="decay_sweep", 

83 updated_by="stigmem:system:decay", 

84 ) 

85 # Append-only retraction log for confidence-floor drops (§24.2.1 c.3) 

86 if conf_ids: 

87 conn.executemany( 

88 "INSERT INTO fact_retractions (id, fact_id, retracted_at, retracted_by) " 

89 "VALUES (?,?,?,?)", 

90 [(str(uuid.uuid4()), fid, now, "stigmem:system:decay") for fid in conf_ids], 

91 ) 

92 # Graph adjacency index (§20.1.2): propagate expiry to entity_edges 

93 from ..recall.graph_index import sync_edge_expiry 

94 

95 sync_edge_expiry(conn, candidates, now) 

96 

97 

98def run_decay_sweep( 

99 ttl_seconds: int | None = None, 

100 min_confidence: float | None = None, 

101 scope: str | None = None, 

102 dry_run: bool = False, 

103 *, 

104 tenant_id: str, 

105) -> dict[str, Any]: 

106 """Mark stale facts as expired by setting valid_until to now. 

107 

108 - TTL decay: non-expiring facts whose timestamp is at or before (now - ttl_seconds). 

109 Passing ttl_seconds=0 expires all non-expiring facts regardless of age. 

110 - Confidence decay: active facts whose confidence is below min_confidence. 

111 - System facts (stigmem: entity/relation, not stigmem://) are never decayed. 

112 - dry_run=True returns counts without writing. 

113 - Candidate selection is scoped to ``tenant_id`` so a caller can only 

114 expire/count facts in its own tenant (no cross-tenant write or count 

115 oracle). 

116 

117 Returns {"scanned": N, "decayed": M, "dry_run": bool}. 

118 """ 

119 now_dt = datetime.now(UTC) 

120 now = now_dt.isoformat() 

121 

122 # Explicit args override settings defaults; settings=0/0.0 means "disabled" 

123 effective_ttl = _resolve_effective_ttl(ttl_seconds) 

124 effective_min_conf = _resolve_effective_min_conf(min_confidence) 

125 

126 ttl_ids: list[str] = [] 

127 conf_ids: list[str] = [] 

128 

129 with db() as conn: 

130 if effective_ttl is not None: 

131 ttl_ids = _select_ttl_candidates(conn, effective_ttl, scope, now_dt, tenant_id) 

132 

133 if effective_min_conf is not None: 

134 conf_ids = _select_confidence_candidates( 

135 conn, effective_min_conf, scope, now, tenant_id 

136 ) 

137 

138 candidates = list({*ttl_ids, *conf_ids}) 

139 

140 if not dry_run and candidates: 

141 _apply_decay(conn, candidates, conf_ids, now) 

142 

143 return { 

144 "scanned": len(candidates), 

145 "decayed": len(candidates) if not dry_run else 0, 

146 "dry_run": dry_run, 

147 }