serve/system1_server.py

"""Local HTTP server speaking TypeSafe's System One contract (POST /v1/systemone) on top of a System1 backend.
Run: python system1_server.py --model <path> [--adapter <safetensors|dir>] [--port 8020] [--model-name system1-9b]
Probabilities come from the same path as scripts/evaluate_system1.py predict (render_messages -> backend.score ->
calibrated_probabilities at T=1). Confidence formulas mirror the Kev reference (Apache-2.0, kev/api.py: choice_confidence,
score_confidence, round_prob), themselves mirroring TypeSafe system-one-adapter 0.2.1 confidence_metrics."""
import argparse
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from kev_convert import kev_record_to_rows
from system1_client import (
    WIDE_CHOICE_KEYS, MAX_PAYLOAD_CHARS, OverlengthError, calibrated_probabilities, render_messages, request_from_row)

MAX_BODY_BYTES = 4 * MAX_PAYLOAD_CHARS  # generous: JSON escaping and multi-question requests; rows are re-bounded per question


class ContractError(ValueError):
    pass


def round_prob(x):
    return round(float(x), 4)


def _normalize(p):
    t = sum(p)
    return [1 / len(p)] * len(p) if t == 0 else [x / t for x in p]


def choice_confidence(p):
    K = len(p)
    return 1.0 if K == 1 else (max(_normalize(p)) - 1 / K) / (1 - 1 / K)


def score_confidence(p):
    L = len(p)
    if L == 1:
        return 1.0
    p = _normalize(p)
    mode = max(range(L), key=p.__getitem__)
    D = sum(abs(i - (L - 1) / 2) for i in range(L)) / L
    return max(0.0, 1.0 - sum(pi * abs(i - mode) for i, pi in enumerate(p)) / D)


def request_to_rows(req):
    """Validate a System One request -> ({qid: row}, {qid: candidate names}). Rows come from kev_convert with a dummy label
    (never rendered), so option text equals what the evaluator and Jev conversion produce."""
    if not isinstance(req, dict):
        raise ContractError("request: expected object")
    if "state" not in req or not isinstance(req["state"], (str, dict, list)):
        raise ContractError("state: required, string, object or array")
    qs = req.get("questions")
    if not isinstance(qs, dict) or not qs:
        raise ContractError("questions: expected nonempty object")
    dummy, rows, names = {}, {}, {}
    for qid, q in qs.items():
        if not isinstance(q, dict):
            raise ContractError(f"{qid}: expected object")
        kind, crit = q.get("type"), q.get("criteria")
        if kind == "choice":
            if not isinstance(crit, (dict, list)) or not 2 <= len(crit) <= len(WIDE_CHOICE_KEYS):
                raise ContractError(f"{qid}: choice requires 2-{len(WIDE_CHOICE_KEYS)} options per choice")
            label = next(iter(crit))
        elif kind == "score":
            if not isinstance(crit, list):
                raise ContractError(f"{qid}: score criteria must be an ordered list")
            label = 0
        elif kind == "noul":
            if crit is not None and not isinstance(crit, dict):
                raise ContractError(f"{qid}: noul criteria must be an object")
            label = True
        else:
            raise ContractError(f"{qid}: unknown question type {kind!r}")
        dummy[qid] = {"type": kind, "instructions": q.get("instructions"), "criteria": crit, "label": label}
    try:
        for row in kev_record_to_rows({"state": req["state"], "questions": dummy}, "serve", "none", "serve", keys=WIDE_CHOICE_KEYS):
            qid = row["meta"]["kev_question_id"]
            rows[qid] = row
            names[qid] = list(row["meta"].get("candidate_ids", {}).values())
    except (ValueError, KeyError, TypeError) as e:
        raise ContractError(str(e))
    return rows, names


def answer_for(row, names, probs):
    kind = row["type"]
    if kind == "noul":
        return {"type": "noul", "noul": round_prob(probs["Y"])}
    if kind == "choice":
        p = [probs[k] for k in row["options"]]
        return {"type": "choice", "choice": names[max(range(len(p)), key=p.__getitem__)], "confidence": round_prob(choice_confidence(p)),
                "probabilities": {n: round_prob(v) for n, v in zip(names, p)}}
    p = [probs[k] for k in row["options"]]
    return {"type": "score", "score": round_prob(sum(i * v for i, v in enumerate(p))),
            "probabilities": {str(i): round_prob(v) for i, v in enumerate(p)}, "confidence": round_prob(score_confidence(p))}


class Service:
    def __init__(self, backend, model_name):
        self.backend, self.model_name, self.lock = backend, model_name, threading.Lock()  # ponytail: one global lock, GPU is serial anyway

    def _tokens(self, text_or_messages):
        tok = getattr(self.backend, "tokenizer", None)
        if tok is None:
            return 0
        if isinstance(text_or_messages, str):
            return len(tok.encode(text_or_messages, add_special_tokens=False))
        prompt = tok.apply_chat_template(text_or_messages, tokenize=False, add_generation_prompt=True, enable_thinking=False)
        return len(tok.encode(prompt, add_special_tokens=False))

    def systemone(self, req):
        start = time.perf_counter()
        rows, names = request_to_rows(req)
        answers, in_tokens = {}, 0
        for qid, row in rows.items():
            try:
                messages = render_messages(request_from_row(row))
            except OverlengthError as e:
                raise ContractError(f"{qid}: payload too large, over the maximum context length: {e}")
            except ValueError as e:
                raise ContractError(f"{qid}: {e}")
            with self.lock:
                try:
                    scores = self.backend.score(messages, list(row["options"]))
                except OverlengthError as e:
                    raise ContractError(f"{qid}: context too large, over the maximum context length: {e}")
                in_tokens += self._tokens(messages)
            if not isinstance(scores, dict) or set(scores) != set(row["options"]):
                raise RuntimeError(f"{qid}: backend returned incomplete scores")
            answers[qid] = answer_for(row, names[qid], calibrated_probabilities(scores, 1.0))
        return {"model": self.model_name, "answers": answers,
                "usage": {"input_tokens": in_tokens, "output_tokens": self._tokens(json.dumps(answers))},
                "latency_ms": round((time.perf_counter() - start) * 1000, 1)}


def make_server(backend, model_name="system1", host="127.0.0.1", port=8020):
    svc = Service(backend, model_name)

    class Handler(BaseHTTPRequestHandler):
        def _send(self, code, body):
            data = json.dumps(body).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def do_GET(self):
            if self.path == "/health":
                self._send(200, {"status": "ok"})
            elif self.path == "/v1/models":
                self._send(200, {"object": "list", "data": [{"id": svc.model_name, "object": "model", "owned_by": "local"}]})
            else:
                self._send(404, {"error": "not found"})

        def do_POST(self):
            if self.path != "/v1/systemone":
                return self._send(404, {"error": "not found"})
            n = int(self.headers.get("Content-Length") or 0)
            if n > MAX_BODY_BYTES:
                return self._send(422, {"error": f"payload too large: {n} bytes"})
            try:
                req = json.loads(self.rfile.read(n))
            except (ValueError, UnicodeDecodeError) as e:
                return self._send(400, {"error": f"malformed JSON: {e}"})
            try:
                self._send(200, svc.systemone(req))
            except ContractError as e:
                self._send(422, {"error": str(e)})
            except Exception as e:
                self._send(500, {"error": f"{type(e).__name__}: {e}"})

        def log_message(self, *a):
            pass

    return ThreadingHTTPServer((host, port), Handler)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--model", required=True)
    ap.add_argument("--adapter")
    ap.add_argument("--port", type=int, default=8020)
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--model-name", default="system1-9b")
    ap.add_argument("--backend", choices=("mlx", "torch"), default="mlx", help="torch: CUDA via transformers")
    a = ap.parse_args()
    import system1_client
    backend = system1_client.TorchBackend if a.backend == "torch" else system1_client.MlxBackend
    server = make_server(backend(a.model, adapter_path=a.adapter), a.model_name, a.host, a.port)
    print(f"serving {a.model_name} on http://{a.host}:{a.port}", flush=True)
    server.serve_forever()


if __name__ == "__main__":
    main()