"""System One shaped inputs; this adapter makes no TypeSafe model claims.""" import json import math MODEL = "decision-kai" MAX_EXPANDED_INPUT_BYTES = 16 * 1024 * 1024 MAX_REQUEST_BYTES = 256 * 1024 MAX_BATCH_REQUEST_BYTES = 2 * 1024 * 1024 MAX_BATCH_DECISIONS = 1024 def content(value, label): if isinstance(value, str): if not value.strip(): raise ValueError(f"{label} must not be empty") return value if isinstance(value, (dict, list)): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False) raise ValueError(f"{label} must be text, an object, or an array") def identifier(value, label, *, max_length=None): if not isinstance(value, str) or not value.strip(): raise ValueError(f"{label} must be a nonempty string") if max_length is not None and len(value) > max_length: raise ValueError(f"{label} must be at most {max_length} characters") return value def _single_records(body, *, model=MODEL): if not isinstance(body, dict) or set(body) - {"state", "questions", "model"}: raise ValueError("Use state, questions, and optional model only") if body.get("model", MODEL) != model: raise ValueError("Request model does not match the selected model") state = content(body.get("state"), "state") questions = body.get("questions") if not isinstance(questions, dict) or not questions: raise ValueError("Provide at least one named question") records = [] for index, (qid, item) in enumerate(questions.items()): identifier(qid, "Question ID") if not isinstance(item, dict) or set(item) - {"type", "instructions", "criteria"}: raise ValueError(f"{qid}: use type, instructions, criteria only") kind = item.get("type") if kind not in ("choice", "noul", "score"): raise ValueError(f"{qid}: type must be choice, noul, or score") q = { "id": qid, "type": kind.capitalize(), "text": content(item.get("instructions"), qid + ".instructions"), } criteria = item.get("criteria") if kind == "choice": if not isinstance(criteria, dict) or not 2 <= len(criteria) <= 255: raise ValueError(f"{qid}: Choice requires 2–255 options") q["options"] = [] for name, description in criteria.items(): identifier(name, "Option name") # Official Choice option names ARE semantic; opaque native IDs are not. text = name if description is None else name + ": " + content(description, qid + ".criteria") q["options"].append({"id": name, "text": text}) elif kind == "score": if not isinstance(criteria, list) or not 2 <= len(criteria) <= 10: raise ValueError(f"{qid}: Score requires 2–10 ordered levels") q["levels"] = [{"id": str(i), "value": i, "text": content(v, qid + ".criteria")} for i, v in enumerate(criteria)] elif criteria is not None: if not isinstance(criteria, dict) or set(criteria) - {"false", "true"}: raise ValueError(f"{qid}: Noul criteria accept false and true only") for key in ("false", "true"): if criteria.get(key) is not None: q[key + "_criterion"] = content(criteria[key], qid + ".criteria." + key) records.append({"id": f"studio:{index}", "state_text": state, "question": q}) return records def _single_response(body, records, predictions, *, model=MODEL, complete_input_tokens=1024, precision="fp32"): if len(predictions) != len(records): raise RuntimeError("Incomplete model response") answers = {} total_tokens = 0 for row, pred in zip(records, predictions): q = row["question"] kind = q["type"].lower() expected_ids = (["no", "yes"] if kind == "noul" else [o["id"] for o in q["options" if kind == "choice" else "levels"]]) if (pred["id"] != row["id"] or pred["question_id"] != q["id"] or pred["candidate_ids"] != expected_ids or pred["type"].lower() != kind or pred["state_tokens_original"] != pred["state_tokens_kept"] or not 1 <= pred["input_tokens"] <= complete_input_tokens): raise RuntimeError("Model response identity or complete-input mismatch") p = pred["probabilities"] if (len(p) != len(expected_ids) or not all(math.isfinite(x) and 0 <= x <= 1 for x in p) or abs(sum(p) - 1) > 2e-5): raise RuntimeError("Invalid probability distribution") total_tokens += pred["input_tokens"] answer = {"type": kind, "input_tokens": pred["input_tokens"]} if kind == "noul": answer["noul"] = p[1] else: answer["probabilities"] = dict(zip(expected_ids, p)) answer["top_probability"] = max(p) if kind == "choice": answer["choice"] = expected_ids[max(range(len(p)), key=p.__getitem__)] else: answer["score"] = sum(i * prob for i, prob in enumerate(p)) answer["legend"] = {str(i): v for i, v in enumerate(body["questions"][q["id"]]["criteria"])} answers[q["id"]] = answer return {"model": model, "answers": answers, "usage": {"input_tokens": total_tokens, "output_tokens": 0}, "profile": {"complete_input_tokens": complete_input_tokens, "precision": precision, "status": "preview", "confidence_definition": None}} def contexts(body, *, model=MODEL): """Validate the whole request before any model work; IDs are bookkeeping only.""" if not isinstance(body, dict) or set(body) - {"state", "states", "questions", "model"}: raise ValueError("Use state OR states, questions, and optional model only") if ("state" in body) == ("states" in body): raise ValueError("Provide exactly one of state or states") if body.get("model", MODEL) != model: raise ValueError("Request model does not match the selected model") limit = MAX_BATCH_REQUEST_BYTES if "states" in body else MAX_REQUEST_BYTES if len(json.dumps(body, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8")) > limit: raise ValueError("Request exceeds the endpoint byte limit; no field is truncated") questions = body.get("questions") if not isinstance(questions, dict) or not questions: raise ValueError("Provide at least one named question") if "state" in body: result = [(None, body["state"])] _expanded_budget(result, questions) return result states = body["states"] if not isinstance(states, list) or not states: raise ValueError("Provide at least one context") if len(states) > MAX_BATCH_DECISIONS or len(questions) > MAX_BATCH_DECISIONS or len(states) * len(questions) > MAX_BATCH_DECISIONS: raise ValueError("A batch may contain at most 1024 states, questions, and decisions") result, seen = [], set() for item in states: if not isinstance(item, dict) or set(item) != {"id", "state"}: raise ValueError("Each context must contain exactly id and state") cid = identifier(item["id"], "Context ID", max_length=128) if cid in seen: raise ValueError("Context IDs must be unique") seen.add(cid) result.append((cid, item["state"])) _expanded_budget(result, questions) return result def _expanded_budget(contexts, questions): # Bound the Cartesian expansion before constructing records. Count limits # are not model limits; repeated long inputs still require bounded work. state_bytes = sum(len(content(state, "state").encode("utf-8")) for _, state in contexts) question_bytes = len(json.dumps(questions, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8")) expanded_bytes = state_bytes * len(questions) + question_bytes * len(contexts) if expanded_bytes > MAX_EXPANDED_INPUT_BYTES: raise ValueError("Expanded context/question input exceeds 16 MiB; split this workload into requests. No field is truncated.") def to_records(body, *, model=MODEL): cc = contexts(body, model=model) if "states" not in body: return _single_records(body, model=model) per_context = [_single_records({"state": state, "questions": body["questions"], "model": model}, model=model) for _, state in cc] records = [] # Question-major order puts different contexts in the same physical B8. for qi in range(len(body["questions"])): for ci in range(len(cc)): row = per_context[ci][qi] opaque = f"studio:batch:{ci}:{qi}" row["id"] = opaque row["question"]["id"] = opaque records.append(row) return records def response_answers(body, result): """Return answers in physical flattened order; reject missing/reordered groups.""" qids = list(body["questions"]) if "states" not in body: if not isinstance(result.get("answers"), dict) or list(result["answers"]) != qids: raise ValueError("Question order mismatch") return list(result["answers"].values()) groups = result.get("results") if not isinstance(groups, list) or len(groups) != len(body["states"]): raise ValueError("Context result count mismatch") for context, group in zip(body["states"], groups): if (not isinstance(group, dict) or set(group) != {"id", "answers", "usage"} or group["id"] != context["id"] or not isinstance(group["answers"], dict) or list(group["answers"]) != qids): raise ValueError("Context/question result identity or order mismatch") return [group["answers"][qid] for qid in qids for group in groups] def to_response(body, records, predictions, *, model=MODEL, complete_input_tokens=1024, precision="fp32"): if records != to_records(body, model=model): raise RuntimeError("Request records changed") if "states" not in body: return _single_response(body, records, predictions, model=model, complete_input_tokens=complete_input_tokens, precision=precision) if len(predictions) != len(records): raise RuntimeError("Incomplete model response") cc = body["states"]; qids = list(body["questions"]); groups = []; total = 0 for ci, context in enumerate(cc): rr, pp = [], [] for qi, qid in enumerate(qids): i = qi * len(cc) + ci; row = records[i]; pred = predictions[i] if pred["id"] != row["id"] or pred["question_id"] != row["question"]["id"]: raise RuntimeError("Flattened context/question identity mismatch") rr.append(dict(row, question=dict(row["question"], id=qid))) pp.append(dict(pred, question_id=qid)) one = _single_response(body, rr, pp, model=model, complete_input_tokens=complete_input_tokens, precision=precision) groups.append({"id": context["id"], "answers": one["answers"], "usage": one["usage"]}) total += one["usage"]["input_tokens"] return {"model": model, "results": groups, "usage": {"input_tokens": total, "output_tokens": 0}, "profile": one["profile"]} def timing_shape(body, records): out = {"questions": len(body["questions"]), "batch_size": min(8, len(records))} if "states" in body: out.update(contexts=len(body["states"]), decisions=len(records)) return out