handler.py

"""
Hugging Face Inference Endpoints custom handler.

Deliberately narrow: this endpoint scores an ALREADY-EXTRACTED feature vector — it
does not accept or walk a raw serialization graph. Graph feature extraction stays
single-sourced in the TS engine (packages/intelligence/src/features.ts); see
docs/huggingface-model-plan.md §0 for why duplicating that traversal here would be
an unacceptable drift risk for a system whose whole premise is "never fabricate."

Expected input (feature order MUST match model.json's `featureNames`):
    {"inputs": {"features": [0.12, 3, 1, 2, 1, 1, 5, 0.0]}}
or a batch:
    {"inputs": [{"features": [...]}, {"features": [...]}]}

Output mirrors the TS AnomalyOutput contract exactly (see scorer.py / model.ts):
    {"score": 0.87, "uncertainty": 0.05, "explanation": "learned anomaly 87% (top: route_rarity)"}
"""

from __future__ import annotations
import os
from typing import Any

from scorer import load_model, score_features


class EndpointHandler:
    def __init__(self, path: str = ""):
        model_path = os.path.join(path, "model.json") if path else "model.json"
        self.model = load_model(model_path)
        self.feature_names: list[str] = self.model.get("featureNames", [])
        self.expected_len = len(self.feature_names) or len(self.model["weights"])

    def _score_one(self, item: dict) -> dict:
        features = item.get("features")
        if not isinstance(features, list) or len(features) != self.expected_len:
            # Never guess: an out-of-contract input abstains completely rather than
            # scoring garbage, mirroring the engine's "abstain when uncertain" rule.
            return {
                "score": 0.0,
                "uncertainty": 1.0,
                "explanation": (
                    f"invalid input: expected a 'features' array of length {self.expected_len} "
                    f"({', '.join(self.feature_names)}) -- model abstains"
                ),
            }
        out = score_features(self.model, [float(x) for x in features])
        return out.to_dict()

    def __call__(self, data: dict[str, Any]) -> list[dict] | dict:
        inputs = data.get("inputs", data)
        if isinstance(inputs, list):
            return [self._score_one(item) for item in inputs]
        return self._score_one(inputs)