diff --git a/plugins/memory/holographic/__init__.py b/plugins/memory/holographic/__init__.py index 927c6cc1a5..9fee7db3f6 100644 --- a/plugins/memory/holographic/__init__.py +++ b/plugins/memory/holographic/__init__.py @@ -1,11 +1,8 @@ -"""hermes-memory-store — holographic memory plugin (MemoryProvider): structured fact storage -with entity resolution, trust scoring, and HRR-based compositional retrieval. -Original plugin by dusterbloom (PR #2351), adapted to the MemoryProvider ABC. - -Config in $HERMES_HOME/config.yaml (profile-scoped) under plugins.hermes-memory-store: - db_path ($HERMES_HOME/memory_store.db), auto_extract (false), default_trust (0.5), - min_trust_threshold (0.3), temporal_decay_half_life (0), hrr_dim (1024), hrr_weight (0.3). -""" +"""hermes-memory-store — holographic memory plugin (MemoryProvider): structured fact storage with entity +resolution, trust scoring, and HRR-based compositional retrieval. Original plugin by dusterbloom (PR #2351). +Config in $HERMES_HOME/config.yaml under plugins.hermes-memory-store: db_path ($HERMES_HOME/memory_store.db), +auto_extract (false), default_trust (0.5), min_trust_threshold (0.3), temporal_decay_half_life (0), +hrr_dim (1024), hrr_weight (0.3).""" from __future__ import annotations @@ -139,8 +136,7 @@ class HolographicMemoryProvider(MemoryProvider): config_path = Path(hermes_home) / "config.yaml" try: import yaml - # Raw read for the write-back round-trip: merged defaults must not be persisted. - from hermes_cli.config import read_user_config_raw + from hermes_cli.config import read_user_config_raw # raw read: merged defaults must not be persisted existing = read_user_config_raw(config_path) existing.setdefault("plugins", {})["hermes-memory-store"] = values with open(config_path, "w", encoding="utf-8") as f: @@ -277,17 +273,12 @@ class HolographicMemoryProvider(MemoryProvider): "fact_feedback": lambda self, a: json.dumps(self._store.record_feedback(int(a["fact_id"]), helpful=a["action"] == "helpful")), } - # -- Auto-extraction (on_session_end) ------------------------------------ - @staticmethod def _harvestable_text(msg: dict): - """User text eligible for extraction, or None. - - Compaction handoff summaries arrive as role="user" and reliably match the decision - patterns; never store the compactor's own output as a durable fact. A merge-into-tail - row holds genuine prior user text BEFORE _MERGED_SUMMARY_DELIMITER (prefixed with the - header) and the summary AFTER it — harvest only the pre-delimiter segment. - """ + """User text eligible for extraction, or None. Compaction handoff summaries arrive as role="user" and + reliably match the decision patterns; never store the compactor's own output as a durable fact. A + merge-into-tail row holds genuine prior user text BEFORE _MERGED_SUMMARY_DELIMITER (prefixed with the + header) and the summary AFTER it — harvest only the pre-delimiter segment.""" # Local import: the compressor module is heavier than this plugin and only needed here. from agent.context_compressor import _MERGED_PRIOR_CONTEXT_HEADER, _MERGED_SUMMARY_DELIMITER, is_compaction_summary_message @@ -306,10 +297,8 @@ class HolographicMemoryProvider(MemoryProvider): def _auto_extract_facts(self, messages: list) -> None: extracted = 0 - for msg in messages: - content = self._harvestable_text(msg) if msg.get("role") == "user" else None - if content is None: - continue + texts = filter(None, (self._harvestable_text(m) for m in messages if m.get("role") == "user")) + for content in texts: for patterns, category in _EXTRACT_CATEGORIES: if any(p.search(content) for p in patterns): try: diff --git a/plugins/memory/holographic/holographic.py b/plugins/memory/holographic/holographic.py index f0422cd7d2..f3736e4363 100644 --- a/plugins/memory/holographic/holographic.py +++ b/plugins/memory/holographic/holographic.py @@ -70,20 +70,16 @@ def encode_text(text: str, dim: int = 1024) -> "np.ndarray": """Bag-of-words bundle of token atoms; empty text -> encode_atom("__hrr_empty__").""" _require_numpy() tokens = [t for t in (tok.strip(".,!?;:\"'()[]{}") for tok in text.lower().split()) if t] - if not tokens: - return encode_atom("__hrr_empty__", dim) - return bundle(*[encode_atom(token, dim) for token in tokens]) + return bundle(*[encode_atom(token, dim) for token in tokens]) if tokens else encode_atom("__hrr_empty__", dim) def encode_fact(content: str, entities: list[str], dim: int = 1024) -> "np.ndarray": """bundle(bind(text, ROLE_CONTENT), bind(entity_i, ROLE_ENTITY)...), so unbind(fact, bind(entity, ROLE_ENTITY)) ≈ content_vector.""" _require_numpy() - role_content = encode_atom("__hrr_role_content__", dim) - role_entity = encode_atom("__hrr_role_entity__", dim) - components = [bind(encode_text(content, dim), role_content)] - components += [bind(encode_atom(entity.lower(), dim), role_entity) for entity in entities] - return bundle(*components) + role_content, role_entity = encode_atom("__hrr_role_content__", dim), encode_atom("__hrr_role_entity__", dim) + return bundle(bind(encode_text(content, dim), role_content), + *[bind(encode_atom(entity.lower(), dim), role_entity) for entity in entities]) def phases_to_bytes(phases: "np.ndarray", dim: int | None = None) -> bytes: @@ -111,43 +107,33 @@ def bytes_to_phases(data: bytes, dim: int | None = None) -> "np.ndarray": prefixed = data.startswith(_FLOAT32_BLOB_PREFIX) f32 = lambda payload: np.frombuffer(payload, dtype=np.float32).astype(np.float64) # noqa: E731 f64 = lambda payload: np.frombuffer(payload, dtype=np.float64).copy() # noqa: E731 - if dim is None: if prefixed: - payload = data[plen:] - if len(payload) % _F32 != 0: - raise ValueError(f"HRR float32 vector blob has invalid payload byte length: {len(payload)}") - return f32(payload) + if (len(data) - plen) % _F32 != 0: + raise ValueError(f"HRR float32 vector blob has invalid payload byte length: {len(data) - plen}") + return f32(data[plen:]) if len(data) % _F64 != 0: raise ValueError(f"HRR legacy vector blob has invalid byte length: {len(data)}") return f64(data) - float32_blob_bytes, float64_bytes = plen + dim * _F32, dim * _F64 collides = float32_blob_bytes == float64_bytes if not collides and prefixed and len(data) == float32_blob_bytes: return f32(data[plen:]) if len(data) == float64_bytes: return f64(data) - if prefixed: - expected = (f"{float64_bytes} (legacy float64)" if collides - else f"{float32_blob_bytes} (prefixed float32) or {float64_bytes} (legacy float64)") - raise ValueError( - f"HRR vector blob has {len(data)} bytes ({len(data) - plen} payload bytes after " - f"the float32 prefix); expected {expected} for dim={dim}" - ) - raise ValueError(f"HRR legacy vector blob has {len(data)} bytes; expected {float64_bytes} (float64) for dim={dim}") + if not prefixed: + raise ValueError(f"HRR legacy vector blob has {len(data)} bytes; expected {float64_bytes} (float64) for dim={dim}") + expected = (f"{float64_bytes} (legacy float64)" if collides + else f"{float32_blob_bytes} (prefixed float32) or {float64_bytes} (legacy float64)") + raise ValueError(f"HRR vector blob has {len(data)} bytes ({len(data) - plen} payload bytes after " + f"the float32 prefix); expected {expected} for dim={dim}") def snr_estimate(dim: int, n_items: int) -> float: """SNR = sqrt(dim / n_items) (inf when empty); warns below 2.0 (n_items > dim/4).""" _require_numpy() - if n_items <= 0: - return float("inf") - snr = math.sqrt(dim / n_items) + snr = math.sqrt(dim / n_items) if n_items > 0 else float("inf") if snr < 2.0: - logger.warning( - "HRR storage near capacity: SNR=%.2f (dim=%d, n_items=%d). " - "Retrieval accuracy may degrade. Consider increasing dim or reducing stored items.", - snr, dim, n_items, - ) + logger.warning("HRR storage near capacity: SNR=%.2f (dim=%d, n_items=%d). " + "Retrieval accuracy may degrade. Consider increasing dim or reducing stored items.", snr, dim, n_items) return snr diff --git a/plugins/memory/holographic/retrieval.py b/plugins/memory/holographic/retrieval.py index ecde0b094b..cc1673a194 100644 --- a/plugins/memory/holographic/retrieval.py +++ b/plugins/memory/holographic/retrieval.py @@ -44,16 +44,9 @@ def _shift(sim: float) -> float: class FactRetriever: """Multi-strategy fact retrieval with trust-weighted scoring.""" - def __init__( - self, - store: MemoryStore, - temporal_decay_half_life: int = 0, # days, 0 = disabled - fts_weight: float = 0.4, jaccard_weight: float = 0.3, hrr_weight: float = 0.3, - hrr_dim: int = 1024, - ): - self.store = store - self.half_life = temporal_decay_half_life - self.hrr_dim = hrr_dim + def __init__(self, store: MemoryStore, temporal_decay_half_life: int = 0, # days, 0 = disabled + fts_weight: float = 0.4, jaccard_weight: float = 0.3, hrr_weight: float = 0.3, hrr_dim: int = 1024): + self.store, self.half_life, self.hrr_dim = store, temporal_decay_half_life, hrr_dim if hrr_weight > 0 and not hrr._HAS_NUMPY: # redistribute weights without numpy fts_weight, jaccard_weight, hrr_weight = 0.6, 0.4, 0.0 self.fts_weight, self.jaccard_weight, self.hrr_weight = fts_weight, jaccard_weight, hrr_weight @@ -86,36 +79,39 @@ class FactRetriever: fact["score"] = relevance * fact["trust_score"] if self.half_life > 0: fact["score"] *= self._temporal_decay(fact.get("updated_at") or fact.get("created_at")) - candidates.sort(key=lambda x: x["score"], reverse=True) - results = candidates[:limit] + results = sorted(candidates, key=lambda x: x["score"], reverse=True)[:limit] for fact in results: fact.pop("hrr_vector", None) # callers expect JSON-serializable dicts return results + def _vector_query(self, fallback: str, category: str | None, limit: int, make_sim: Callable[[], Callable]) -> list[dict]: + """Rank every fact vector (optionally per category) with the sim fn built by make_sim(); FTS5 fallback + when no vectors exist. make_sim runs after the rows check so role atoms are encoded once, never for fallback.""" + rows = self._vector_rows(category) + if not rows: + return self.search(fallback, category=category, limit=limit) + return self._rank_by_vector(rows, make_sim(), limit) + def probe(self, entity: str, category: str | None = None, limit: int = 10) -> list[dict]: """Compositional entity query: unbind bind(entity, ROLE_ENTITY) from the category bank (or each fact vector) to find facts where the entity plays a structural role. Not keyword search. Falls back to FTS5 without numpy.""" if not hrr._HAS_NUMPY: return self.search(entity, category=category, limit=limit) - role_entity = self._atom(_ROLE_ENTITY) - probe_key = hrr.bind(self._atom(entity.lower()), role_entity) + probe_key = hrr.bind(self._atom(entity.lower()), self._atom(_ROLE_ENTITY)) if category: # category bank first, then individual fact vectors bank_row = self.store._conn.execute("SELECT vector FROM memory_banks WHERE bank_name = ?", (f"cat:{category}",)).fetchone() if bank_row: extracted = hrr.unbind(self._phases(bank_row["vector"]), probe_key) return self._rank_by_vector(self._vector_rows(category), lambda _f, fact_vec: hrr.similarity(extracted, fact_vec), limit) - rows = self._vector_rows(category) - if not rows: - return self.search(entity, category=category, limit=limit) - role_content = self._atom(_ROLE_CONTENT) # loop-invariant — encode once - def _sim(fact: dict, fact_vec) -> float: + def make_sim(): + role_content = self._atom(_ROLE_CONTENT) # Does unbinding the probe key leave the fact's content signal? - residual = hrr.unbind(fact_vec, probe_key) - return hrr.similarity(residual, hrr.bind(hrr.encode_text(fact["content"], self.hrr_dim), role_content)) + return lambda fact, fact_vec: hrr.similarity( + hrr.unbind(fact_vec, probe_key), hrr.bind(hrr.encode_text(fact["content"], self.hrr_dim), role_content)) - return self._rank_by_vector(rows, _sim, limit) + return self._vector_query(entity, category, limit, make_sim) def related(self, entity: str, category: str | None = None, limit: int = 10) -> list[dict]: """Facts structurally connected to an entity (shared context), not just facts @@ -123,17 +119,13 @@ class FactRetriever: if not hrr._HAS_NUMPY: return self.search(entity, category=category, limit=limit) entity_vec = self._atom(entity.lower()) # bare atom, not role-bound: ANY structural match - rows = self._vector_rows(category) - if not rows: - return self.search(entity, category=category, limit=limit) - role_entity, role_content = self._atom(_ROLE_ENTITY), self._atom(_ROLE_CONTENT) # encode once - def _sim(fact: dict, fact_vec) -> float: + def make_sim(): + roles = (self._atom(_ROLE_ENTITY), self._atom(_ROLE_CONTENT)) # A residual similar to ANY role vector means the entity plays a structural role. - residual = hrr.unbind(fact_vec, entity_vec) - return max(hrr.similarity(residual, role_entity), hrr.similarity(residual, role_content)) + return lambda _f, fact_vec: max(hrr.similarity(hrr.unbind(fact_vec, entity_vec), role) for role in roles) - return self._rank_by_vector(rows, _sim, limit) + return self._vector_query(entity, category, limit, make_sim) def reason(self, entities: list[str], category: str | None = None, limit: int = 10) -> list[dict]: """Multi-entity compositional query (vector-space JOIN): facts where ALL entities @@ -142,16 +134,13 @@ class FactRetriever: return self.search(" ".join(entities), category=category, limit=limit) role_entity = self._atom(_ROLE_ENTITY) probe_keys = [hrr.bind(self._atom(entity.lower()), role_entity) for entity in entities] - rows = self._vector_rows(category) - if not rows: - return self.search(" ".join(entities), category=category, limit=limit) - role_content = self._atom(_ROLE_CONTENT) - def _sim(fact: dict, fact_vec) -> float: + def make_sim(): + role_content = self._atom(_ROLE_CONTENT) # AND semantics via min: high only if EVERY entity is structurally present. - return min(hrr.similarity(hrr.unbind(fact_vec, key), role_content) for key in probe_keys) + return lambda _f, fact_vec: min(hrr.similarity(hrr.unbind(fact_vec, key), role_content) for key in probe_keys) - return self._rank_by_vector(rows, _sim, limit) + return self._vector_query(" ".join(entities), category, limit, make_sim) def contradict(self, category: str | None = None, threshold: float = 0.3, limit: int = 10) -> list[dict]: """Memory hygiene: pairs of facts that share entities (same subject) but have low @@ -163,41 +152,33 @@ class FactRetriever: return [] if len(rows) > 500: # O(n²) guard: only compare the most recently updated facts rows = sorted(rows, key=lambda r: r["updated_at"] or r["created_at"], reverse=True)[:500] - facts = [dict(r) for r in rows] - for fact in facts: + facts = [] # (public dict, lower-cased entity names, phase vector) + for row in rows: + fact = dict(row) entity_rows = self.store._conn.execute( "SELECT e.name FROM entities e JOIN fact_entities fe ON fe.entity_id = e.entity_id WHERE fe.fact_id = ?", (fact["fact_id"],), ).fetchall() - fact["_entities"] = {r["name"].lower() for r in entity_rows} - fact["_vec"] = self._phases(fact.pop("hrr_vector")) - - def _public(fact: dict) -> dict: - return {k: v for k, v in fact.items() if k not in ("_entities", "_vec")} - + facts.append((fact, {r["name"].lower() for r in entity_rows}, self._phases(fact.pop("hrr_vector")))) contradictions = [] - for i, f1 in enumerate(facts): - for f2 in facts[i + 1:]: - ents1, ents2 = f1["_entities"], f2["_entities"] + for i, (f1, ents1, vec1) in enumerate(facts): + for f2, ents2, vec2 in facts[i + 1:]: if not ents1 or not ents2: continue entity_overlap = len(ents1 & ents2) / len(ents1 | ents2) if entity_overlap < 0.3: continue # not enough shared subject to be contradictory - content_sim = hrr.similarity(f1["_vec"], f2["_vec"]) + content_sim = hrr.similarity(vec1, vec2) contradiction_score = entity_overlap * (1.0 - _shift(content_sim)) # high overlap + low similarity if contradiction_score >= threshold: contradictions.append({ - "fact_a": _public(f1), "fact_b": _public(f2), + "fact_a": f1, "fact_b": f2, "entity_overlap": round(entity_overlap, 3), "content_similarity": round(content_sim, 3), "contradiction_score": round(contradiction_score, 3), "shared_entities": sorted(ents1 & ents2), }) - contradictions.sort(key=lambda x: x["contradiction_score"], reverse=True) - return contradictions[:limit] - - # -- Vector scoring helpers ----------------------------------------------- + return sorted(contradictions, key=lambda x: x["contradiction_score"], reverse=True)[:limit] def _vector_rows(self, category: str | None, columns: str = _FACT_COLUMNS + ", hrr_vector") -> list: """All facts that carry an HRR vector, optionally filtered by category.""" @@ -209,21 +190,14 @@ class FactRetriever: scored = [dict(row) for row in rows] for fact in scored: fact["score"] = _shift(sim_fn(fact, self._phases(fact.pop("hrr_vector")))) * fact["trust_score"] - scored.sort(key=lambda x: x["score"], reverse=True) - return scored[:limit] - - # -- FTS / lexical helpers ------------------------------------------------ + return sorted(scored, key=lambda x: x["score"], reverse=True)[:limit] def _fts_candidates(self, query: str, category: str | None, min_trust: float, limit: int) -> list[dict]: """Raw FTS5 MATCH candidates with rank normalized to [0, 1] as 'fts_rank'.""" category_clause = "AND f.category = ? " if category else "" params = [self._sanitize_fts_query(query)] + ([category] if category else []) + [min_trust, limit] - sql = ( - "SELECT f.*, facts_fts.rank as fts_rank_raw FROM facts_fts " - "JOIN facts f ON f.fact_id = facts_fts.rowid " - f"WHERE facts_fts MATCH ? {category_clause}AND f.trust_score >= ? " - "ORDER BY facts_fts.rank LIMIT ?" - ) + sql = ("SELECT f.*, facts_fts.rank as fts_rank_raw FROM facts_fts JOIN facts f ON f.fact_id = facts_fts.rowid " + f"WHERE facts_fts MATCH ? {category_clause}AND f.trust_score >= ? ORDER BY facts_fts.rank LIMIT ?") try: results = [dict(row) for row in self.store._conn.execute(sql, params).fetchall()] except Exception: @@ -243,12 +217,9 @@ class FactRetriever: @staticmethod def _sanitize_fts_query(query: str) -> str: - """Natural-language query -> FTS5-safe OR expression of quoted tokens. - - FTS5 AND-joins a multi-word MATCH by default, which tanks recall on prose. Drops - stopwords and <2-char tokens, strips FTS5 operator chars, and phrase-quotes each - survivor. If nothing survives, returns the raw query (zero results, not a SQL error). - """ + """Natural-language query -> FTS5-safe OR expression of quoted tokens. FTS5 AND-joins a multi-word + MATCH by default, which tanks recall on prose: drop stopwords and <2-char tokens, strip FTS5 operator + chars, phrase-quote each survivor. If nothing survives, return the raw query (zero results, not a SQL error).""" if not query: return "" tokens = [f'"{c}"' for c in (raw.strip(_PUNCT).translate(_FTS_OPERATORS) for raw in query.lower().split()) @@ -265,9 +236,7 @@ class FactRetriever: if not self.half_life or not timestamp_str: return 1.0 try: - ts = timestamp_str - if isinstance(ts, str): - ts = datetime.fromisoformat(ts.replace("Z", "+00:00")) + ts = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00")) if isinstance(timestamp_str, str) else timestamp_str if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) age_days = (datetime.now(timezone.utc) - ts).total_seconds() / 86400 diff --git a/plugins/memory/holographic/store.py b/plugins/memory/holographic/store.py index 4c1cc01dca..6b52bfadc4 100644 --- a/plugins/memory/holographic/store.py +++ b/plugins/memory/holographic/store.py @@ -99,12 +99,10 @@ def _clamp_trust(value: float) -> float: class MemoryStore: """SQLite-backed fact store with entity resolution and trust scoring. - Process-wide shared connection registry: SQLite allows one writer at a time and - several providers coexist per process (main agent + every delegate_task subagent), - so all instances for the same database share ONE connection and ONE re-entrant - lock — writes are fully serialized and "database is locked" is impossible. - Refcounted: closing one instance never tears the connection out from under a sibling. - """ + Process-wide shared connection registry: SQLite allows one writer at a time and several providers + coexist per process (main agent + every delegate_task subagent), so all instances for the same database + share ONE connection and ONE re-entrant lock — writes are fully serialized and "database is locked" is + impossible. Refcounted: closing one instance never tears the connection out from under a sibling.""" _shared: dict = {} _shared_guard = threading.Lock() @@ -115,9 +113,7 @@ class MemoryStore: db_path = str(get_hermes_home() / "memory_store.db") self.db_path = Path(db_path).expanduser() self.db_path.parent.mkdir(parents=True, exist_ok=True) - self.default_trust = _clamp_trust(default_trust) - self.hrr_dim = hrr_dim - self._hrr_available = hrr._HAS_NUMPY + self.default_trust, self.hrr_dim, self._hrr_available = _clamp_trust(default_trust), hrr_dim, hrr._HAS_NUMPY try: # resolve() so symlinked/relative paths to the same file share ONE connection self._key = str(self.db_path.resolve()) except OSError: @@ -155,8 +151,6 @@ class MemoryStore: self._conn.commit() return cur - # -- Public API ----------------------------------------------------------- - def add_fact(self, content: str, category: str = "general", tags: str = "") -> int: """Insert a fact and return its fact_id; on duplicate content (UNIQUE) return the existing fact_id untouched. Links extracted entities and rebuilds the category bank.""" @@ -181,14 +175,12 @@ class MemoryStore: row = self._one("SELECT fact_id, trust_score FROM facts WHERE fact_id = ?", (fact_id,)) if row is None: return False - changes = [(col, val) for col, val in ( - ("content", content.strip() if content is not None else None), - ("tags", tags), - ("category", category), - ("trust_score", _clamp_trust(row["trust_score"] + trust_delta) if trust_delta is not None else None), - ) if val is not None] - assignments = ", ".join(["updated_at = CURRENT_TIMESTAMP"] + [f"{col} = ?" for col, _ in changes]) - self._write(f"UPDATE facts SET {assignments} WHERE fact_id = ?", [val for _, val in changes] + [fact_id]) + changes = {col: val for col, val in { + "content": content.strip() if content is not None else None, "tags": tags, "category": category, + "trust_score": _clamp_trust(row["trust_score"] + trust_delta) if trust_delta is not None else None, + }.items() if val is not None} + assignments = ", ".join(["updated_at = CURRENT_TIMESTAMP"] + [f"{col} = ?" for col in changes]) + self._write(f"UPDATE facts SET {assignments} WHERE fact_id = ?", [*changes.values(), fact_id]) if content is not None: # re-extract entities and recompute the HRR vector self._conn.execute("DELETE FROM fact_entities WHERE fact_id = ?", (fact_id,)) self._link_entities(fact_id, content) @@ -227,13 +219,10 @@ class MemoryStore: raise KeyError(f"fact_id {fact_id} not found") old_trust: float = row["trust_score"] new_trust = _clamp_trust(old_trust + (_HELPFUL_DELTA if helpful else _UNHELPFUL_DELTA)) - helpful_increment = 1 if helpful else 0 + increment = 1 if helpful else 0 self._write("UPDATE facts SET trust_score = ?, helpful_count = helpful_count + ?, " - "updated_at = CURRENT_TIMESTAMP WHERE fact_id = ?", (new_trust, helpful_increment, fact_id)) - return {"fact_id": fact_id, "old_trust": old_trust, "new_trust": new_trust, - "helpful_count": row["helpful_count"] + helpful_increment} - - # -- Entity / HRR helpers ------------------------------------------------- + "updated_at = CURRENT_TIMESTAMP WHERE fact_id = ?", (new_trust, increment, fact_id)) + return {"fact_id": fact_id, "old_trust": old_trust, "new_trust": new_trust, "helpful_count": row["helpful_count"] + increment} def _extract_entities(self, text: str) -> list[str]: """Regex entity candidates (see the pattern table), deduplicated case-insensitively in first-seen order.""" @@ -287,16 +276,12 @@ class MemoryStore: (bank_name, hrr.phases_to_bytes(bank_vector), self.hrr_dim, len(rows)), ) - # -- Lifecycle ------------------------------------------------------------ - @classmethod def release_all_under(cls, directory: "str | Path") -> int: """Force-close every shared connection whose database lives under ``directory``; returns the count. - - close() is refcount-driven, so a live holder (e.g. an agent's provider) keeps a profile's SQLite - handle open, which on Windows makes rmtree of the profile fail. The directory is going away, so - later use by a stale holder is expected to fail. - """ + close() is refcount-driven, so a live holder (e.g. an agent's provider) keeps a profile's SQLite handle + open, which on Windows makes rmtree of the profile fail. The directory is going away, so later use by a + stale holder is expected to fail.""" root = os.path.normcase(str(Path(directory).expanduser().resolve())) + os.sep with cls._shared_guard: doomed = [key for key in cls._shared if os.path.normcase(key).startswith(root)]