src/routing.py

"""
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