Self-healing fuzzy trail + /ui/trails endpoint

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+marketing/videos/*.mp4 filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -4,3 +4,4 @@ __pycache__/
 .env
 *.log
 .DS_Store
+marketing/videos/

core/signal_registry.py

@@ -1,54 +1,196 @@
-import json
-import os
+import re
 import time
-from typing import Dict, List, Any, Optional
+from difflib import SequenceMatcher
+from typing import Dict, List, Any, Optional, Tuple
 from app.logger import logger
 
+
 class SignalStream:
     def __init__(self, name: str, source_type: str, data: Any, metadata: Dict[str, Any]):
         self.name = name
-        self.source_type = source_type # 'rss', 'mcp', 'context', 'log'
+        self.source_type = source_type  # 'rss', 'mcp', 'context', 'log'
         self.data = data
         self.metadata = metadata
         self.timestamp = time.time()
 
+
 class SignalRegistry:
     """
     The Signal Mesh: A central hub where data sources broadcast signals.
-    Agents (Antennae) tune into these based on their spatial/functional mapping.
+    Agents (Antennae) tune in based on keywords.
+
+    Miss handling — self-healing fuzzy trail:
+      1. Score unmapped keyword against all live frequencies (SequenceMatcher + token Jaccard)
+      2. If best score >= TRAIL_THRESHOLD: acknowledge gap, write persistent trail
+         (keyword → [best_freq, ...token-overlap neighbors])
+      3. Branch outward from best match to capture related frequencies
+      4. Future calls with that keyword hit the trail as a fast-path — no re-scoring
+      5. Gap metadata rides along in the returned signal so the caller can inspect it
     """
     _instance = None
+    TRAIL_THRESHOLD  = 0.45   # min combined score to write a trail
+    BRANCH_THRESHOLD = 0.30   # min token-overlap ratio to include a neighbor
 
     def __new__(cls):
         if cls._instance is None:
-            cls._instance = super(SignalRegistry, cls).__new__(cls)
+            cls._instance = super().__new__(cls)
             cls._instance.streams: Dict[str, List[SignalStream]] = {}
+            cls._instance.fuzzy_trails: Dict[str, List[str]] = {}
+            cls._instance.trail_gaps: Dict[str, Dict[str, float]] = {}
         return cls._instance
 
-    def broadcast(self, name: str, source_type: str, data: Any, metadata: Optional[Dict[str, Any]] = None):
-        """Broadcast a signal into the mesh."""
+    # ── Broadcast ─────────────────────────────────────────────────────────────
+
+    def broadcast(self, name: str, source_type: str, data: Any,
+                  metadata: Optional[Dict[str, Any]] = None):
         if name not in self.streams:
             self.streams[name] = []
-        
-        stream = SignalStream(name, source_type, data, metadata or {})
-        self.streams[name].append(stream)
-        # Keep last 100 signals per frequency
+        self.streams[name].append(SignalStream(name, source_type, data, metadata or {}))
         self.streams[name] = self.streams[name][-100:]
         logger.info(f"📡 Signal Mesh: Broadcasting stream '{name}' ({source_type})")
 
+    # ── Matching helpers ───────────────────────────────────────────────────────
+
+    @staticmethod
+    def _tokens(s: str) -> set:
+        return set(re.split(r'[_\-\s]+', s.lower())) - {''}
+
+    @staticmethod
+    def _similarity(a: str, b: str) -> float:
+        return SequenceMatcher(None, a.lower(), b.lower()).ratio()
+
+    def _fast_match(self, freq_name: str, kw: str) -> bool:
+        kl = kw.lower()
+        fl = freq_name.lower()
+        if kl in fl:
+            return True
+        if self._tokens(kl) & self._tokens(fl):
+            return True
+        # camelCase decomposition
+        camel = set(re.sub(r'([A-Z])', r' \1', kw).lower().split())
+        return bool(camel & self._tokens(fl))
+
+    def _score_all(self, keyword: str) -> List[Tuple[str, float]]:
+        kl = keyword.lower()
+        results = []
+        for name in self.streams:
+            tok_a   = self._tokens(kl)
+            tok_b   = self._tokens(name.lower())
+            union   = tok_a | tok_b
+            jaccard = len(tok_a & tok_b) / len(union) if union else 0.0
+            score   = 0.6 * self._similarity(kl, name) + 0.4 * jaccard
+            results.append((name, score))
+        return sorted(results, key=lambda x: x[1], reverse=True)
+
+    def _branch_from(self, anchor: str) -> List[str]:
+        anchor_tokens = self._tokens(anchor.lower())
+        neighbors = []
+        for name in self.streams:
+            if name == anchor:
+                continue
+            overlap = len(anchor_tokens & self._tokens(name.lower()))
+            if overlap and overlap / max(len(anchor_tokens), 1) >= self.BRANCH_THRESHOLD:
+                neighbors.append(name)
+        return neighbors
+
+    # ── Trail writer ───────────────────────────────────────────────────────────
+
+    def _write_trail(self, keyword: str, scores: List[Tuple[str, float]]) -> List[str]:
+        best_freq, best_score = scores[0]
+        if best_score < self.TRAIL_THRESHOLD:
+            logger.warning(
+                f"🕸  No trail for '{keyword}' — "
+                f"best match '{best_freq}' @ {best_score:.2f} (threshold {self.TRAIL_THRESHOLD})"
+            )
+            return []
+
+        branches = self._branch_from(best_freq)
+        trail    = [best_freq] + branches
+        self.fuzzy_trails[keyword] = trail
+        self.trail_gaps[keyword]   = {f: round(s, 3) for f, s in scores[:5]}
+        logger.info(
+            f"🕸  Trail written: '{keyword}' → {trail}  "
+            f"(anchor '{best_freq}' @ {best_score:.2f}, {len(branches)} branch(es))"
+        )
+        return trail
+
+    # ── tune_in ────────────────────────────────────────────────────────────────
+
     def tune_in(self, keywords: List[str]) -> List[Dict[str, Any]]:
-        """Antennae use this to catch signals matching their frequency (keywords)."""
-        caught_signals = []
-        for name, buffer in self.streams.items():
-            # Match signals to agent antennae keywords or name
-            if any(key.lower() in name.lower() for key in keywords):
-                for signal in buffer:
-                    caught_signals.append({
-                        "name": signal.name,
-                        "type": signal.source_type,
-                        "content": signal.data,
-                        "age": time.time() - signal.timestamp
-                    })
-        return caught_signals
+        """
+        Match priority per keyword:
+          1. Learned fuzzy trail fast-path
+          2. Direct fast match (substring / token / camelCase)
+          3. Miss → score all → write trail if above threshold → return with _trail metadata
+        """
+        matched: Dict[str, Optional[str]] = {}   # freq → trail_kw|None
+        gap_report: Dict[str, Dict] = {}
+
+        for kw in keywords:
+            # 1. Trail fast-path
+            if kw in self.fuzzy_trails:
+                for freq in self.fuzzy_trails[kw]:
+                    if freq in self.streams:
+                        matched.setdefault(freq, kw)
+                continue
+
+            # 2. Direct match
+            direct = False
+            for name in self.streams:
+                if self._fast_match(name, kw):
+                    matched.setdefault(name, None)
+                    direct = True
+            if direct:
+                continue
+
+            # 3. Miss → benchmark → write trail
+            scores = self._score_all(kw)
+            trail  = self._write_trail(kw, scores)
+            for freq in trail:
+                if freq in self.streams:
+                    matched.setdefault(freq, kw)
+            if trail:
+                gap_report[kw] = {
+                    "bridged_to": trail[0],
+                    "confidence": round(scores[0][1], 3),
+                    "branches":   trail[1:],
+                    "top5":       {f: round(s, 3) for f, s in scores[:5]},
+                }
+
+        # Collect signals
+        caught: List[Dict[str, Any]] = []
+        seen: set = set()
+        for freq_name, via_kw in matched.items():
+            for signal in self.streams.get(freq_name, []):
+                sig_id = (signal.name, signal.timestamp)
+                if sig_id in seen:
+                    continue
+                seen.add(sig_id)
+                entry: Dict[str, Any] = {
+                    "name":    signal.name,
+                    "type":    signal.source_type,
+                    "content": signal.data,
+                    "age":     time.time() - signal.timestamp,
+                }
+                if via_kw and via_kw in gap_report:
+                    entry["_trail"] = gap_report[via_kw]
+                caught.append(entry)
+
+        return caught
+
+    # ── Introspection ──────────────────────────────────────────────────────────
+
+    def get_trails(self) -> Dict[str, Any]:
+        return {
+            kw: {"frequencies": freqs, "gap_scores": self.trail_gaps.get(kw, {})}
+            for kw, freqs in self.fuzzy_trails.items()
+        }
+
+    def clear_trail(self, keyword: str) -> bool:
+        removed = keyword in self.fuzzy_trails
+        self.fuzzy_trails.pop(keyword, None)
+        self.trail_gaps.pop(keyword, None)
+        return removed
+
 
 signal_registry = SignalRegistry()

signalmesh_api.py

@@ -357,6 +357,21 @@ async def ui_hf(req: HfInjectReq): return await _do_hf_inject(req.repo_id, req.r
 @app.post("/ui/arxiv_inject")
 async def ui_arxiv(req: ArxivInjectReq): return await _do_arxiv_inject(req.arxiv_id)
 
+@app.get("/ui/trails")
+def ui_trails():
+    """Return all learned fuzzy trails — keyword → frequencies + gap scores."""
+    return registry.get_trails()
+
+@app.delete("/ui/trails/{keyword}")
+def ui_clear_trail(keyword: str):
+    removed = registry.clear_trail(keyword)
+    return {"removed": removed, "keyword": keyword}
+
+@app.get("/api/trails")
+def api_trails(x_signalmesh_key: Optional[str] = Header(default=None)):
+    _check_key(x_signalmesh_key)
+    return registry.get_trails()
+
 # ── Dashboard (v3 — clean bright build) ──────────────────────────────────────
 _HTML = r"""<!DOCTYPE html>
 <html lang="en">