Upload handler.py with huggingface_hub

handler.py

@@ -0,0 +1,53 @@
+"""
+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)