""" RoutingMatrix - JSON skill routing with precedence and typed keys. Deterministic skill index (from matrix), hot-swappable LoRA adapter paths. Back-compatible with .skill files via load_legacy_skill(). """ import hashlib import json import re from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Optional @dataclass class RouteDecision: skill: Optional[str] = None token: Optional[str] = None index: Optional[int] = None adapter: Optional[str] = None score: float = 0.0 def is_skill(self) -> bool: return self.skill is not None def deterministic_index(name: str, num_slots: int = 64) -> int: """Stable skill->index mapping (replaces runtime hash(), which varies per process).""" return int(hashlib.sha256(name.encode("utf-8")).hexdigest(), 16) % num_slots class RoutingMatrix: def __init__(self, matrix_path: Optional[str] = None): self.skills: List[Dict] = [] self.default: Dict = {} if matrix_path: self.load(matrix_path) def load(self, matrix_path: str): data = json.loads(Path(matrix_path).read_text(encoding="utf-8")) self.skills = data.get("skills", []) self.default = data.get("default", {}) return len(self.skills) def load_legacy_skill(self, skill_path: str) -> Dict: """Adopt a .skill file into the matrix (keeps old runtime working).""" data = json.loads(Path(skill_path).read_text(encoding="utf-8")) entry = { "name": data["name"], "token": data["token"], "index": deterministic_index(data["name"]), "priority": 10, "patterns": [ {"type": "keyword", "value": p} for p in data.get("trigger_patterns", []) ], } self.skills.append(entry) return entry def _score(self, text: str, entry: Dict) -> float: low = text.lower() score = 0.0 for pat in entry.get("patterns", []): value = pat.get("value", "") ptype = pat.get("type", "regex") if ptype == "regex": if re.search(value, low): score += 1.0 elif ptype == "keyword": if value.lower() in low: score += 0.8 return score def route(self, text: str) -> RouteDecision: """Highest-scoring skill wins; ties broken by priority.""" best = RouteDecision() for entry in self.skills: score = self._score(text, entry) if score == 0: continue entry_prio = entry.get("priority", 0) best_prio = 0 if best.score == 0 else self._priority_of(best.skill) if score > best.score or (score == best.score and entry_prio > best_prio): best = RouteDecision( skill=entry["name"], token=entry.get("token"), index=entry.get("index", deterministic_index(entry["name"])), adapter=entry.get("adapter"), score=score, ) return best def _priority_of(self, name: Optional[str]) -> int: if not name: return 0 for e in self.skills: if e.get("name") == name: return e.get("priority", 0) return 0