feat: add pi-agent extension for hexus vector memory

.gitignore

@@ -46,3 +46,6 @@ CLAUDE.md
 AGENTS.md
 .planning/
 .claude/
+
+# Node
+node_modules/

benchmarks/bench.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 """Benchmark script for hexus.
 
 Run inside the docker container:
@@ -166,13 +165,12 @@ def make_text(i: int) -> str:
     print(f"  Cross-agent (None) sees: {len(rows_all)} rows")
 
     # Cleanup
-    with store._get_pool().connection() as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                "DELETE FROM memory_entries WHERE agent_identity IN (%s, %s)",
-                (agent, agent2),
-            )
-            conn.commit()
+    with store._get_pool().connection() as conn, conn.cursor() as cur:
+        cur.execute(
+            "DELETE FROM memory_entries WHERE agent_identity IN (%s, %s)",
+            (agent, agent2),
+        )
+        conn.commit()
     store.close()
 
 

hexus/ccr/cache.py

@@ -1,17 +1,16 @@
 # Forked from andreab67/hermes-memory-pgvector (BSD-3-Clause)
 import threading
-from typing import Dict, Optional
 
 
 class CCRCache:
     """Thread-safe in-memory cache mapping memory_id (int) to compressed text (str)."""
 
     def __init__(self, maxsize: int = 1000):
         self._maxsize = maxsize
-        self._cache: Dict[int, str] = {}
+        self._cache: dict[int, str] = {}
         self._lock = threading.Lock()
 
-    def get(self, memory_id: int) -> Optional[str]:
+    def get(self, memory_id: int) -> str | None:
         with self._lock:
             if memory_id in self._cache:
                 val = self._cache.pop(memory_id)

hexus/embed.py

@@ -21,7 +21,6 @@
 import logging
 import urllib.error
 import urllib.request
-from typing import List, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -49,11 +48,11 @@ class EmbeddingError(Exception):
 def embed(
     text: str,
     *,
-    base_url: Optional[str] = None,
-    model: Optional[str] = None,
+    base_url: str | None = None,
+    model: str | None = None,
     timeout: float = 10.0,
     expected_dim: int = EXPECTED_DIM,
-) -> List[float]:
+) -> list[float]:
     """Return an embedding for `text`.
 
     Dispatch:
@@ -88,9 +87,11 @@ def embed(
         # path (e.g. a CI env that just runs unit tests against a mock
         # endpoint).
         from .embedder import (
+            DEFAULT_MODEL,
             get_default_embedder,
+        )
+        from .embedder import (
             EmbedderError as _LocalErr,
-            DEFAULT_MODEL,
         )
 
         embedder = get_default_embedder(model_name=model or DEFAULT_MODEL)
@@ -129,7 +130,7 @@ def embed(
 
 def _post(
     url: str, body: dict, *, timeout: float, expected_dim: int, extract
-) -> List[float]:
+) -> list[float]:
     data = json.dumps(body).encode("utf-8")
     req = urllib.request.Request(
         url,
@@ -159,7 +160,7 @@ def _post(
     return vec
 
 
-def to_hexus_literal(vec: List[float]) -> str:
+def to_hexus_literal(vec: list[float]) -> str:
     """Render a Python list of floats as a hexus input literal.
 
     psycopg can also handle this via type adapters, but the literal form

hexus/embedder.py

@@ -33,7 +33,6 @@
 os.environ.setdefault("USER", "agy")
 import threading
 from dataclasses import dataclass, replace
-from typing import Dict, List, Optional, Tuple
 
 logger = logging.getLogger(__name__)
 
@@ -96,7 +95,7 @@ class EmbedStats:
     tokens_dropped: int = 0  # approx tokens lost to truncation (Σ tc-max_seq)
     max_tokens_seen: int = 0  # largest single-text token count observed
 
-    def as_dict(self) -> Dict[str, int]:
+    def as_dict(self) -> dict[str, int]:
         return {
             "texts_embedded": self.texts_embedded,
             "texts_over_limit": self.texts_over_limit,
@@ -108,7 +107,7 @@ def as_dict(self) -> Dict[str, int]:
         }
 
 
-def _resolve_long_text_mode(mode: Optional[str]) -> str:
+def _resolve_long_text_mode(mode: str | None) -> str:
     """Resolve the configured mode: explicit arg > env var > default.
 
     An unrecognised value falls back to the default with a warning rather
@@ -156,9 +155,9 @@ def __init__(
         self,
         model_name: str = DEFAULT_MODEL,
         *,
-        cache_dir: Optional[str] = None,
+        cache_dir: str | None = None,
         device: str = "cpu",
-        long_text_mode: Optional[str] = None,
+        long_text_mode: str | None = None,
     ):
         self._model_name = model_name
         self._cache_dir = cache_dir
@@ -201,8 +200,10 @@ def dim(self) -> int:
             # transformer config. Fall through to the constant if not.
             try:
                 return int(self._model.get_sentence_embedding_dimension())
-            except Exception:  # noqa: BLE001
-                pass
+            except Exception as exc:  # noqa: BLE001
+                logger.debug(
+                    "hexus embedder: failed to get embedding dimension: %s", exc
+                )
         return DEFAULT_DIM if self._model_name == DEFAULT_MODEL else 0
 
     @property
@@ -229,7 +230,7 @@ def reset_stats(self) -> None:
         with self._stats_lock:
             self._stats = EmbedStats()
 
-    def embed(self, texts: List[str]) -> List[List[float]]:
+    def embed(self, texts: list[str]) -> list[list[float]]:
         """Embed a list of texts → list of float vectors.
 
         Empty / whitespace-only inputs are silently dropped (returned
@@ -266,7 +267,7 @@ def embed(self, texts: List[str]) -> List[List[float]]:
                 convert_to_numpy=True,
                 show_progress_bar=False,
             )
-        except Exception as exc:  # noqa: BLE001 — fail-soft, surface in logs
+        except Exception as exc:
             raise EmbedderError(f"local embed failed: {exc}") from exc
 
         # Reassemble one vector per original input (chunk windows collapse
@@ -291,8 +292,8 @@ def embed(self, texts: List[str]) -> List[List[float]]:
     # -- Long-input handling (issue #7) -------------------------------------
 
     def _plan_encode(
-        self, texts: List[str], model
-    ) -> Tuple[List[str], List[Tuple[int, int, Optional[List[float]]]]]:
+        self, texts: list[str], model
+    ) -> tuple[list[str], list[tuple[int, int, list[float] | None]]]:
         """Turn `texts` into (`pieces` to encode, `plan` to reassemble them).
 
         Each plan entry is `(start, count, weights)`:
@@ -313,8 +314,8 @@ def _plan_encode(
         )
 
         mode = self._long_text_mode
-        pieces: List[str] = []
-        plan: List[Tuple[int, int, Optional[List[float]]]] = []
+        pieces: list[str] = []
+        plan: list[tuple[int, int, list[float] | None]] = []
 
         for i, text in enumerate(texts):
             with self._stats_lock:
@@ -329,8 +330,7 @@ def _plan_encode(
             # Over the limit — count it (in every mode) and record the peak.
             with self._stats_lock:
                 self._stats.texts_over_limit += 1
-                if tc > self._stats.max_tokens_seen:
-                    self._stats.max_tokens_seen = tc
+                self._stats.max_tokens_seen = max(self._stats.max_tokens_seen, tc)
                 over_count = self._stats.texts_over_limit
 
             if mode == LONG_TEXT_MODE_CHUNK:
@@ -361,8 +361,8 @@ def _plan_encode(
         return pieces, plan
 
     def _assemble(
-        self, raw, plan: List[Tuple[int, int, Optional[List[float]]]]
-    ) -> List[List[float]]:
+        self, raw, plan: list[tuple[int, int, list[float] | None]]
+    ) -> list[list[float]]:
         """Collapse encoded `pieces` back to one vector per original input.
 
         Single-piece entries are returned verbatim (byte-identical to the
@@ -379,7 +379,7 @@ def _assemble(
 
         import numpy as np
 
-        vectors: List[List[float]] = []
+        vectors: list[list[float]] = []
         for start, count, weights in plan:
             if count == 1:
                 vectors.append(raw[start].tolist())
@@ -406,7 +406,7 @@ def _resolve_max_seq(model) -> int:
         except (TypeError, ValueError):
             return 0
 
-    def _token_counts(self, texts: List[str], tokenizer) -> Optional[List[int]]:
+    def _token_counts(self, texts: list[str], tokenizer) -> list[int] | None:
         """True (untruncated) token count per text, or None if unavailable.
 
         `verbose=False` suppresses HuggingFace's "sequence longer than model
@@ -427,7 +427,7 @@ def _token_counts(self, texts: List[str], tokenizer) -> Optional[List[int]]:
             )
             return None
 
-    def _chunk_text(self, text: str, tokenizer, max_seq: int) -> List[Tuple[str, int]]:
+    def _chunk_text(self, text: str, tokenizer, max_seq: int) -> list[tuple[str, int]]:
         """Split `text` into overlapping token windows → [(chunk_text, n_tokens)].
 
         We reserve room for the special tokens the tokenizer re-adds when each
@@ -451,7 +451,7 @@ def _chunk_text(self, text: str, tokenizer, max_seq: int) -> List[Tuple[str, int
         if len(ids) <= window:
             return [(text, len(ids))]
 
-        chunks: List[Tuple[str, int]] = []
+        chunks: list[tuple[str, int]] = []
         for start in range(0, len(ids), stride):
             window_ids = ids[start : start + window]
             if not window_ids:
@@ -550,7 +550,7 @@ def _load_model(self):
                     self._device,
                 )
                 return self._model
-            except Exception as exc:  # noqa: BLE001
+            except Exception as exc:
                 self._load_failed = True
                 raise EmbedderError(
                     f"failed to load sentence-transformers model {self._model_name}: {exc}"
@@ -564,17 +564,17 @@ def _load_model(self):
 # Caching is keyed on (model_name, cache_dir, device) so a request for a
 # different model returns a different embedder (mostly relevant for tests
 # — production uses one model). The dict is small in practice.
-_singletons: dict[tuple[str, Optional[str], str, str], "LocalBertEmbedder"] = {}
+_singletons: dict[tuple[str, str | None, str, str], LocalBertEmbedder] = {}
 _singleton_lock = threading.Lock()
 
 
 def get_default_embedder(
     model_name: str = DEFAULT_MODEL,
     *,
-    cache_dir: Optional[str] = None,
-    device: Optional[str] = None,
-    long_text_mode: Optional[str] = None,
-) -> "LocalBertEmbedder":
+    cache_dir: str | None = None,
+    device: str | None = None,
+    long_text_mode: str | None = None,
+) -> LocalBertEmbedder:
     """Return the process-wide default embedder for these args, constructing
     it on first call. Subsequent calls with the same (model_name, cache_dir,
     device, long_text_mode) return the same instance.
@@ -587,7 +587,6 @@ def get_default_embedder(
     part of the cache key so a request for a different mode returns a
     distinct embedder rather than silently reusing another mode's instance.
     """
-    global _singletons
     if device is None:
         device = os.environ.get("HEXUS_EMBED_DEVICE", "cpu")
     mode = _resolve_long_text_mode(long_text_mode)

hexus/entity_extractor.py

@@ -1,5 +1,4 @@
 import re
-from typing import List, Dict
 
 DEFAULT_PATTERNS = {
     "url": r'https?://[^\s<>"]+',
@@ -14,12 +13,12 @@
 
 
 class EntityExtractor:
-    def __init__(self, patterns: Dict[str, str] = None, enabled: bool = True):
+    def __init__(self, patterns: dict[str, str] | None = None, enabled: bool = True):
         self.enabled = enabled
         self.patterns = {**DEFAULT_PATTERNS, **(patterns or {})}
         self._compiled = {t: re.compile(p) for t, p in self.patterns.items()}
 
-    def extract_entities(self, text: str) -> List[Dict[str, str]]:
+    def extract_entities(self, text: str) -> list[dict[str, str]]:
         if not self.enabled or not text:
             return []
 

hexus/pipeline/router.py

@@ -1,6 +1,6 @@
 # Forked from andreab67/hermes-memory-pgvector (BSD-3-Clause)
-import re
 import json
+import re
 
 
 class ContentRouter:
@@ -63,7 +63,7 @@ def _compress_json(self, text: str) -> str:
                 return f"[Compressed JSON Object] keys: {', '.join(keys)}\nSample: {json.dumps(truncated)}"
             elif isinstance(data, list):
                 return f"[Compressed JSON Array] length: {len(data)}\nFirst item: {json.dumps(data[0]) if data else 'empty'}"
-        except Exception:
+        except Exception:  # noqa: BLE001, S110
             pass
         return text[: self.threshold_chars] + "\n... [Truncated JSON]"
 
@@ -98,9 +98,9 @@ def _compress_code(self, text: str) -> str:
         lines = text.splitlines()
         compressed_lines = []
         for line in lines:
-            if re.match(r"^\s*(def|class|import|from|async\s+def)\b", line):
-                compressed_lines.append(line)
-            elif re.match(r"^\s*#.*", line) and len(compressed_lines) < 10:
+            if re.match(r"^\s*(def|class|import|from|async\s+def)\b", line) or (
+                re.match(r"^\s*#.*", line) and len(compressed_lines) < 10
+            ):
                 compressed_lines.append(line)
 
         if len(compressed_lines) < 3: