Coverage for astrocyte/policy/signal

1"""Signal quality policies — deduplication and noisy-bank detection.

3All functions are sync (Rust migration candidates).

4See docs/_design/policy-layer.md section 3.

5"""

7from __future__ import annotations

9import math

10import re

11import time

12from collections import Counter, deque

13from collections.abc import Callable, Iterable

15# Explicit negators. Deliberately a closed list: this guard targets the

16# measurable failure (a statement and its negation embed as near-duplicates),

17# not contradiction in general. Antonym reversals ("enable"/"disable") carry no

18# negator and are NOT caught here — they also score lower (0.870 measured vs

19# 0.920 for "allergic"/"not allergic"), so they sit further below threshold.

20_NEGATORS = frozenset({"not", "no", "never", "neither", "nor", "without", "cannot", "none", "nobody", "nothing"})

21_WORD = re.compile(r"\b\w+(?:'\w+)*\b")

24def _tokens(text: str) -> list[str]:

25 return _WORD.findall((text or "").casefold().replace("\u2019", "'"))

28def _is_negator(token: str) -> bool:

29 return token in _NEGATORS or token.endswith("n't")

32def differs_by_negation(a: str, b: str) -> bool:

33 """True when the token difference between ``a`` and ``b`` includes a negator.

35 Embedding similarity cannot see the scope of a negation. Measured with

36 bge-small: "The user is allergic to peanuts." vs "The user is not allergic

37 to peanuts." scores 0.920 cosine, against 0.396 for an unrelated pair. A

38 cosine-only dedup would treat the reversal as the duplicate and drop it.

40 Why "differs by" rather than "either side contains": dedup operates on

41 whole conversation chunks, and a long chunk almost always contains a

42 negator somewhere. Requiring exact equality whenever one appears (the rule

43 engrim uses for short curated records) would disable dedup for re-ingested

44 conversations that differ only in a timestamp. Asking whether the

45 *difference* contains a negator targets exactly the reversal case and

46 leaves ordinary re-ingest dedup intact.

48 Multiset difference, so "not ... not" vs "not" still counts.

49 """

50 ca, cb = Counter(_tokens(a)), Counter(_tokens(b))

51 diff = (ca - cb) + (cb - ca)

52 return any(_is_negator(t) for t in diff)

55def cosine_similarity(a: list[float], b: list[float]) -> float:

56 """Compute cosine similarity between two vectors.

58 Returns value in [-1.0, 1.0]. Returns 0.0 for zero vectors.

59 Sync, pure computation — Rust migration candidate.

60 """

61 if len(a) != len(b):

62 raise ValueError(f"Vector dimension mismatch: {len(a)} != {len(b)}")

64 dot = sum(x * y for x, y in zip(a, b))

65 norm_a = math.sqrt(sum(x * x for x in a))

66 norm_b = math.sqrt(sum(x * x for x in b))

68 if norm_a == 0.0 or norm_b == 0.0:

69 return 0.0

71 return dot / (norm_a * norm_b)

74class DedupDetector:

75 """Detect near-duplicate content via embedding similarity.

77 Stores recent embeddings per bank for comparison.

78 Sync, self-contained — Rust migration candidate.

79 """

81 _MAX_BANKS = 1000

83 def __init__(self, similarity_threshold: float = 0.95, max_cache_per_bank: int = 1000) -> None:

84 self.threshold = similarity_threshold

85 self.max_cache = max_cache_per_bank

86 # bank_id -> list of (memory_id, embedding, text). ``text`` is None for

87 # entries added without it; those fall back to cosine-only.

88 self._cache: dict[str, list[tuple[str, list[float], str | None]]] = {}

89 #: How often the negation guard overrode a cosine match. Observable so

90 #: the guard's effect can be measured rather than assumed.

91 self.negation_overrides: int = 0

93 def _touch_bank(self, bank_id: str) -> None:

94 """Move bank to end of dict (most recently used) for LRU eviction."""

95 if bank_id in self._cache:

96 self._cache[bank_id] = self._cache.pop(bank_id)

98 def is_duplicate(

99 self,

100 bank_id: str,

101 embedding: list[float],

102 threshold_override: float | None = None,

103 text: str | None = None,

104 ) -> tuple[bool, float]:

105 """Check if embedding is a near-duplicate of cached content.

106

107 ``threshold_override`` lets a per-call MIP DedupSpec.threshold take

108 precedence over the instance-level default. When ``None``, the

109 configured ``self.threshold`` is used.

110

111 When ``text`` is given and the cached entry has text too, a cosine match

112 is vetoed if the two differ by a negator (see ``differs_by_negation``).

113 The asymmetry is deliberate: the default dedup action *drops* the new

114 chunk, so a false duplicate silently loses a reversed fact, while a

115 false non-duplicate only stores a redundant row. Bias toward keeping.

116

117 Returns (is_dup, max_similarity).

118 """

119 entries = self._cache.get(bank_id, [])

120 if entries:

121 self._touch_bank(bank_id)

122 scored = ((cosine_similarity(embedding, cached_emb), cached_text) for _, cached_emb, cached_text in entries)

123 return self.matches(scored, threshold_override=threshold_override, text=text)

124

125 def find_duplicate(

126 self,

127 bank_id: str,

128 embedding: list[float],

129 threshold_override: float | None = None,

130 text: str | None = None,

131 ) -> str | None:

132 """Like :meth:`is_duplicate`, but returns the cached memory id the

133 embedding duplicates (the first match, under the same threshold and

134 negation guard), or ``None``."""

135 entries = self._cache.get(bank_id, [])

136 if entries:

137 self._touch_bank(bank_id)

138 for memory_id, cached_emb, cached_text in entries:

139 pair = [(cosine_similarity(embedding, cached_emb), cached_text)]

140 if self.matches(pair, threshold_override=threshold_override, text=text)[0]:

141 return memory_id

142 return None

143

144 def matches(

145 self,

146 candidates: Iterable[tuple[float, str | None]],

147 threshold_override: float | None = None,

148 text: str | None = None,

149 ) -> tuple[bool, float]:

150 """Apply the duplicate rule to ``(similarity, text)`` pairs scored elsewhere.

151

152 Shared by the cache scan above and by the retain pipeline's check

153 against the vector store's nearest neighbours, so both apply the same

154 threshold and the same negation guard. Returns (is_dup, max_similarity).

155 """

156 threshold = threshold_override if threshold_override is not None else self.threshold

157 max_sim = 0.0

158 for sim, other_text in candidates:

159 max_sim = max(max_sim, sim)

160 if sim >= threshold:

161 if text is not None and other_text is not None and differs_by_negation(text, other_text):

162 self.negation_overrides += 1

163 continue # a reversal, not a duplicate — keep looking

164 return True, sim

165 return False, max_sim

166

167 def add(self, bank_id: str, memory_id: str, embedding: list[float], text: str | None = None) -> None:

168 """Add an embedding to the cache for future dedup checks.

169

170 Pass ``text`` to enable the negation guard against this entry.

171 """

172 if bank_id not in self._cache:

173 if len(self._cache) >= self._MAX_BANKS:

174 # Evict least-recently-used bank (first key in insertion-ordered dict)

175 lru_bank = next(iter(self._cache))

176 del self._cache[lru_bank]

177 self._cache[bank_id] = []

178 self._touch_bank(bank_id)

179

180 entries = self._cache[bank_id]

181 entries.append((memory_id, embedding, text))

182

183 # Evict oldest if over capacity

184 if len(entries) > self.max_cache:

185 self._cache[bank_id] = entries[-self.max_cache :]

186

187 def clear_bank(self, bank_id: str) -> None:

188 """Clear cache for a bank."""

189 self._cache.pop(bank_id, None)

190

191 def remove(self, bank_id: str, memory_id: str) -> bool:

192 """Drop a single ``(memory_id, embedding)`` entry from the cache.

193

194 Called from the forget pipeline so a re-retain of similar content

195 after forget produces a fresh row instead of silently dedup'ing

196 against the cached embedding of the now-deleted memory.

197

198 Returns ``True`` if an entry was removed, ``False`` otherwise.

199 Idempotent — multiple forgets of the same id are safe.

200 """

201 entries = self._cache.get(bank_id)

202 if not entries:

203 return False

204

205 before = len(entries)

206 self._cache[bank_id] = [e for e in entries if e[0] != memory_id]

207

208 # Reclaim the bank slot entirely if it became empty.

209 if not self._cache[bank_id]:

210 del self._cache[bank_id]

211

212 return len(self._cache.get(bank_id, [])) < before

213

214

215class NoisyBankDetector:

216 """Flags banks whose recent retains look like noise (policy-layer.md §3.3).

217

218 Three signals, each from retains this process has seen for the bank:

219

220 - ``retain_spike``: the last minute's retain count exceeds

221 ``retain_spike_multiplier`` × the per-minute average before it (a runaway

222 agent loop);

223 - ``short_content``: the average length of recent retains is below

224 ``min_avg_content_length`` characters (junk);

225 - ``high_dedup_rate``: more than ``max_dedup_rate`` of recent retains were

226 near-duplicates (redundant).

227

228 Samples expire after ``HORIZON_SECONDS``, so a flag clears once the bank

229 behaves — or, when the caller throttles or rejects flagged retains (which

230 are then never recorded), once the window has passed. In-memory and

231 per-process like ``DedupDetector``; thresholds are passed to ``check`` so

232 each bank can use its own ``noisy_bank`` settings.

233 """

234

235 _MAX_BANKS = 1000

236 #: Most recent retains the content-length and dedup-rate signals look at.

237 SAMPLE_WINDOW = 50

238 #: Below this many samples those two signals stay quiet: a new bank's first

239 #: few short notes are not a pattern.

240 MIN_SAMPLES = 20

241 #: Samples older than this are forgotten.

242 HORIZON_SECONDS = 3600.0

243 #: A spike needs at least this many retains in the last minute...

244 MIN_BURST = 10

245 #: ...and at least this much earlier history in the window to compare with.

246 MIN_BASELINE_SECONDS = 300.0

247 _MAX_EVENTS_PER_BANK = 10_000

248

249 def __init__(self, clock: Callable[[], float] = time.monotonic) -> None:

250 self._clock = clock

251 # bank_id -> (timestamp, content_length, deduplicated), oldest first.

252 self._events: dict[str, deque[tuple[float, int, bool]]] = {}

253 # bank_id -> reasons it was flagged for at its last check.

254 self._flagged: dict[str, tuple[str, ...]] = {}

255

256 def record(self, bank_id: str, content_length: int, deduplicated: bool) -> None:

257 """Record one processed retain (stored or deduplicated)."""

258 events = self._events.pop(bank_id, None)

259 if events is None:

260 events = deque(maxlen=self._MAX_EVENTS_PER_BANK)

261 self._events[bank_id] = events # re-insert: most recently used last

262 events.append((self._clock(), content_length, deduplicated))

263 while len(self._events) > self._MAX_BANKS:

264 evicted = next(iter(self._events))

265 del self._events[evicted]

266 self._flagged.pop(evicted, None)

267

268 def check(

269 self,

270 bank_id: str,

271 *,

272 retain_spike_multiplier: float,

273 min_avg_content_length: int,

274 max_dedup_rate: float,

275 ) -> tuple[tuple[str, ...], bool]:

276 """Return ``(reasons, changed)``: why the bank is flagged now (empty

277 when it is not), and whether that differs from its previous check — so

278 callers can log transitions instead of every flagged retain."""

279 reasons = self._reasons(bank_id, retain_spike_multiplier, min_avg_content_length, max_dedup_rate)

280 changed = reasons != self._flagged.get(bank_id, ())

281 if reasons:

282 self._flagged[bank_id] = reasons

283 else:

284 self._flagged.pop(bank_id, None)

285 return reasons, changed

286

287 def _reasons(

288 self,

289 bank_id: str,

290 spike_multiplier: float,

291 min_avg_length: int,

292 max_dedup_rate: float,

293 ) -> tuple[str, ...]:

294 events = self._events.get(bank_id)

295 if not events:

296 return ()

297 now = self._clock()

298 while events and events[0][0] < now - self.HORIZON_SECONDS:

299 events.popleft()

300 if not events:

301 return ()

302 reasons: list[str] = []

303

304 minute_ago = now - 60.0

305 last_minute = sum(1 for ts, _, _ in events if ts > minute_ago)

306 earlier = len(events) - last_minute

307 baseline_seconds = minute_ago - events[0][0]

308 if last_minute >= self.MIN_BURST and earlier and baseline_seconds >= self.MIN_BASELINE_SECONDS:

309 # Floor of one retain a minute: a quiet bank waking up to a handful

310 # of writes is not a runaway loop.

311 baseline_per_minute = max(earlier / (baseline_seconds / 60.0), 1.0)

312 if last_minute > spike_multiplier * baseline_per_minute:

313 reasons.append("retain_spike")

314

315 recent = list(events)[-self.SAMPLE_WINDOW :]

316 if len(recent) >= self.MIN_SAMPLES:

317 if sum(length for _, length, _ in recent) / len(recent) < min_avg_length:

318 reasons.append("short_content")

319 if sum(1 for _, _, dup in recent if dup) / len(recent) > max_dedup_rate:

320 reasons.append("high_dedup_rate")

321 return tuple(reasons)

Coverage for astrocyte/policy/signal_quality.py: 99%

137 statements