pivot_serving_schema.py

"""Pivot-Alpha / Jev-aligned typed decision schema (English).

Jev: unstructured state in → typed probabilistic decisions out.
We are NOT a free-text generator. Each call returns structured decisions
the host program can wire into a workflow.
"""
from __future__ import annotations

from typing import Any, Literal, Optional

Primitive = Literal["choice", "noul", "score"]


def build_typed_decision(
    *,
    decision_id: str,
    primitive: Primitive,
    options: list[str],
    index: int,
    probs: list[float],
    description: Optional[str] = None,
) -> dict[str, Any]:
    if len(options) != len(probs):
        raise ValueError("options/probs length mismatch")
    if not (0 <= index < len(options)):
        raise ValueError("index out of range")
    # named map for program consumption
    prob_map = {str(opt): float(p) for opt, p in zip(options, probs)}
    value = options[index]
    out: dict[str, Any] = {
        "id": decision_id,
        "primitive": primitive,
        "description": description,
        "options": list(options),
        "index": int(index),
        "value": value,
        "probs": prob_map,
        "prob_vector": [float(p) for p in probs],
        "confidence": float(max(probs) if probs else 0.0),
    }
    if primitive == "noul":
        # binary probabilistic decision (yes-mass = prob of first true-like option if present)
        true_aliases = {"true", "yes", "y", "1"}
        true_idx = next((i for i, o in enumerate(options) if str(o).lower() in true_aliases), index)
        out["p_true"] = float(probs[true_idx])
    if primitive == "score":
        # expected score if options are ordered numeric levels; else keep categorical
        try:
            levels = [float(o) for o in options]
            out["expected"] = float(sum(l * p for l, p in zip(levels, probs)))
        except ValueError:
            out["expected"] = None
    return out


def build_response(
    *,
    state: str,
    decisions: list[dict[str, Any]],
    model_id: str = "Pivot-Alpha",
) -> dict[str, Any]:
    return {
        "model": model_id,
        "contract": "unstructured_state_in__typed_probabilistic_decisions_out",
        "state": state,
        "decisions": decisions,
        "schema_version": "pivot-alpha-v1",
    }