Upload folder using huggingface_hub

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+tokenizer.json filter=lfs diff=lfs merge=lfs -text

README.md

@@ -0,0 +1,59 @@
+---
+license: apache-2.0
+base_model: Qwen/Qwen3.5-0.8B-Base
+language: [en]
+pipeline_tag: text-classification
+tags: [decision-model, calibrated, structured-output, multi-task, system-one, one-pass]
+---
+
+# decider-0.8b: typed decisions with calibrated probabilities in one forward pass
+
+The small sibling of [decider-2b](https://huggingface.co/Mapika/decider-2b): a language model that does not generate text. It reads
+a **state** (a string or any JSON value) and a set of **typed questions** and returns a probability distribution for every question
+from a single forward pass: Choice (2-255 options, optionally described), Score (2-10 described levels), Noul (probability of yes).
+No decoding, no parsing, no output outside the options you defined. Same code, same wire format (`POST /v1/systemone`, TypeSafe
+Jev's format), same training recipe as the 2B: one epoch of `scripts/train.sh full` from `Qwen/Qwen3.5-0.8B-Base` over the full
+mixture (1.47M examples, 455M tokens, 4.5 h on one GH200). Code: https://github.com/Mapika/decider
+
+```python
+# pip install git+https://github.com/Mapika/decider
+from decider.infer import Decider
+d = Decider("Mapika/decider-0.8b")
+d.system_one(
+    {"ticket": "I was charged twice for order A-104. Please refund the duplicate."},
+    {"team": {"type": "choice", "instructions": "Which team should handle this?",
+              "criteria": {"billing": "Charges, invoices, refunds", "technical": "Bugs, outages", "other": None}},
+     "refund_requested": {"type": "noul", "instructions": "Does the customer ask for a refund?"},
+     "frustration": {"type": "score", "instructions": "How frustrated is the customer?", "criteria": ["calm", "frustrated", "very frustrated"]}})
+```
+
+## How it compares with the 2B
+
+Same 93 public tasks, same protocol, one temperature fitted on in-task data (it came out at 1.03 for both: the recipe calibrates
+at every size). "Held-out" means no example of that dataset was trained on.
+
+| | decider-0.8b | decider-2b (same single-run recipe) |
+|---|---|---|
+| in-task accuracy / ECE, 69 tasks | 0.776 / 0.032 | 0.809 / 0.030 |
+| held-out accuracy / ECE, 24 tasks | 0.707 / 0.096 | 0.739 / 0.086 |
+| schema-first layout (the cacheable one), in-task / held-out accuracy | 0.770 / 0.699 | 0.790 / 0.707 |
+| teacher-written custom questions, held-out domains: noul / choice / score | 0.94 / 0.95 / 0.81 | 0.98 / 0.97 / 0.83 |
+| terse-bucket routing, held-out domains: generic / specific / catch-all | 0.87 / 0.93 / 0.84 | 0.91 / 0.94 / 0.92 |
+| JSON state, one of 16 / 64 records named by path (64 with indices written in) | 0.58 / 0.52 (0.62) | 0.59 / 0.53 (0.63) |
+| all 64 / 50 / 70 / 219 labels at once: HWU64, TREC-fine, DBpedia L2, L3 (held-out) | 0.81 / 0.63 / 0.66 / 0.83 | 0.85 / 0.76 / 0.72 / 0.86 |
+| QuALITY, whole article (5-8k tokens) | 0.63 | 0.68 |
+| isolated Score levels vs listwise (teacher-written, held-out domains) | 0.83 vs 0.82, fits sum to 1.02 | 0.84 vs 0.84 |
+| hand-written battery: generic option right / catch-all right | 0.95 / 0.95 | 0.90 / 0.85 |
+
+What the smaller model gives up is knowledge, not the decision format: the largest drops are TruthfulQA (0.41 vs 0.55), OpenBookQA
+(0.67 vs 0.81), HellaSwag (0.76 vs 0.88) and ARC (0.74 vs 0.86), and wide label sets that need fine distinctions (TREC-fine). Routing,
+classification, yes/no judgments and JSON lookups on short states are within one to four points of the 2B. It is a weaker player:
+Pong and Breakout stay at the scripted teacher's level, CliffWalking fails (it walks off the cliff), held-out Freeway scores 0.
+The regression set runs about 1.5x faster than on the 2B; bf16 weights are 1.5 GB.
+
+## Limitations
+
+Those of decider-2b, more so: a small model without reasoning; English only; rules written into the question ("fill if empty,
+otherwise skip") are not followed reliably, so state the decision as a plain question with described options; knowledge-heavy
+multiple choice is close to the base model; calibration is measured on public datasets and teacher-labelled probes, not on your
+traffic. The teacher-written training data comes from Qwen3.5-27B and carries its biases.

chat_template.jinja

@@ -0,0 +1,154 @@
+{%- set image_count = namespace(value=0) %}
+{%- set video_count = namespace(value=0) %}
+{%- macro render_content(content, do_vision_count, is_system_content=false) %}
+    {%- if content is string %}
+        {{- content }}
+    {%- elif content is iterable and content is not mapping %}
+        {%- for item in content %}
+            {%- if 'image' in item or 'image_url' in item or item.type == 'image' %}
+                {%- if is_system_content %}
+                    {{- raise_exception('System message cannot contain images.') }}
+                {%- endif %}
+                {%- if do_vision_count %}
+                    {%- set image_count.value = image_count.value + 1 %}
+                {%- endif %}
+                {%- if add_vision_id %}
+                    {{- 'Picture ' ~ image_count.value ~ ': ' }}
+                {%- endif %}
+                {{- '<|vision_start|><|image_pad|><|vision_end|>' }}
+            {%- elif 'video' in item or item.type == 'video' %}
+                {%- if is_system_content %}
+                    {{- raise_exception('System message cannot contain videos.') }}
+                {%- endif %}
+                {%- if do_vision_count %}
+                    {%- set video_count.value = video_count.value + 1 %}
+                {%- endif %}
+                {%- if add_vision_id %}
+                    {{- 'Video ' ~ video_count.value ~ ': ' }}
+                {%- endif %}
+                {{- '<|vision_start|><|video_pad|><|vision_end|>' }}
+            {%- elif 'text' in item %}
+                {{- item.text }}
+            {%- else %}
+                {{- raise_exception('Unexpected item type in content.') }}
+            {%- endif %}
+        {%- endfor %}
+    {%- elif content is none or content is undefined %}
+        {{- '' }}
+    {%- else %}
+        {{- raise_exception('Unexpected content type.') }}
+    {%- endif %}
+{%- endmacro %}
+{%- if not messages %}
+    {{- raise_exception('No messages provided.') }}
+{%- endif %}
+{%- if tools and tools is iterable and tools is not mapping %}
+    {{- '<|im_start|>system\n' }}
+    {{- "# Tools\n\nYou have access to the following functions:\n\n<tools>" }}
+    {%- for tool in tools %}
+        {{- "\n" }}
+        {{- tool | tojson }}
+    {%- endfor %}
+    {{- "\n</tools>" }}
+    {{- '\n\nIf you choose to call a function ONLY reply in the following format with NO suffix:\n\n<tool_call>\n<function=example_function_name>\n<parameter=example_parameter_1>\nvalue_1\n</parameter>\n<parameter=example_parameter_2>\nThis is the value for the second parameter\nthat can span\nmultiple lines\n</parameter>\n</function>\n</tool_call>\n\n<IMPORTANT>\nReminder:\n- Function calls MUST follow the specified format: an inner <function=...></function> block must be nested within <tool_call></tool_call> XML tags\n- Required parameters MUST be specified\n- You may provide optional reasoning for your function call in natural language BEFORE the function call, but NOT after\n- If there is no function call available, answer the question like normal with your current knowledge and do not tell the user about function calls\n</IMPORTANT>' }}
+    {%- if messages[0].role == 'system' %}
+        {%- set content = render_content(messages[0].content, false, true)|trim %}
+        {%- if content %}
+            {{- '\n\n' + content }}
+        {%- endif %}
+    {%- endif %}
+    {{- '<|im_end|>\n' }}
+{%- else %}
+    {%- if messages[0].role == 'system' %}
+        {%- set content = render_content(messages[0].content, false, true)|trim %}
+        {{- '<|im_start|>system\n' + content + '<|im_end|>\n' }}
+    {%- endif %}
+{%- endif %}
+{%- set ns = namespace(multi_step_tool=true, last_query_index=messages|length - 1) %}
+{%- for message in messages[::-1] %}
+    {%- set index = (messages|length - 1) - loop.index0 %}
+    {%- if ns.multi_step_tool and message.role == "user" %}
+        {%- set content = render_content(message.content, false)|trim %}
+        {%- if not(content.startswith('<tool_response>') and content.endswith('</tool_response>')) %}
+            {%- set ns.multi_step_tool = false %}
+            {%- set ns.last_query_index = index %}
+        {%- endif %}
+    {%- endif %}
+{%- endfor %}
+{%- if ns.multi_step_tool %}
+    {{- raise_exception('No user query found in messages.') }}
+{%- endif %}
+{%- for message in messages %}
+    {%- set content = render_content(message.content, true)|trim %}
+    {%- if message.role == "system" %}
+        {%- if not loop.first %}
+            {{- raise_exception('System message must be at the beginning.') }}
+        {%- endif %}
+    {%- elif message.role == "user" %}
+        {{- '<|im_start|>' + message.role + '\n' + content + '<|im_end|>' + '\n' }}
+    {%- elif message.role == "assistant" %}
+        {%- set reasoning_content = '' %}
+        {%- if message.reasoning_content is string %}
+            {%- set reasoning_content = message.reasoning_content %}
+        {%- else %}
+            {%- if '</think>' in content %}
+                {%- set reasoning_content = content.split('</think>')[0].rstrip('\n').split('<think>')[-1].lstrip('\n') %}
+                {%- set content = content.split('</think>')[-1].lstrip('\n') %}
+            {%- endif %}
+        {%- endif %}
+        {%- set reasoning_content = reasoning_content|trim %}
+        {%- if loop.index0 > ns.last_query_index %}
+            {{- '<|im_start|>' + message.role + '\n<think>\n' + reasoning_content + '\n</think>\n\n' + content }}
+        {%- else %}
+            {{- '<|im_start|>' + message.role + '\n' + content }}
+        {%- endif %}
+        {%- if message.tool_calls and message.tool_calls is iterable and message.tool_calls is not mapping %}
+            {%- for tool_call in message.tool_calls %}
+                {%- if tool_call.function is defined %}
+                    {%- set tool_call = tool_call.function %}
+                {%- endif %}
+                {%- if loop.first %}
+                    {%- if content|trim %}
+                        {{- '\n\n<tool_call>\n<function=' + tool_call.name + '>\n' }}
+                    {%- else %}
+                        {{- '<tool_call>\n<function=' + tool_call.name + '>\n' }}
+                    {%- endif %}
+                {%- else %}
+                    {{- '\n<tool_call>\n<function=' + tool_call.name + '>\n' }}
+                {%- endif %}
+                {%- if tool_call.arguments is defined %}
+                    {%- for args_name, args_value in tool_call.arguments|items %}
+                        {{- '<parameter=' + args_name + '>\n' }}
+                        {%- set args_value = args_value | tojson | safe if args_value is mapping or (args_value is sequence and args_value is not string) else args_value | string %}
+                        {{- args_value }}
+                        {{- '\n</parameter>\n' }}
+                    {%- endfor %}
+                {%- endif %}
+                {{- '</function>\n</tool_call>' }}
+            {%- endfor %}
+        {%- endif %}
+        {{- '<|im_end|>\n' }}
+    {%- elif message.role == "tool" %}
+        {%- if loop.previtem and loop.previtem.role != "tool" %}
+            {{- '<|im_start|>user' }}
+        {%- endif %}
+        {{- '\n<tool_response>\n' }}
+        {{- content }}
+        {{- '\n</tool_response>' }}
+        {%- if not loop.last and loop.nextitem.role != "tool" %}
+            {{- '<|im_end|>\n' }}
+        {%- elif loop.last %}
+            {{- '<|im_end|>\n' }}
+        {%- endif %}
+    {%- else %}
+        {{- raise_exception('Unexpected message role.') }}
+    {%- endif %}
+{%- endfor %}
+{%- if add_generation_prompt %}
+    {{- '<|im_start|>assistant\n' }}
+    {%- if enable_thinking is defined and enable_thinking is true %}
+        {{- '<think>\n' }}
+    {%- else %}
+        {{- '<think>\n\n</think>\n\n' }}
+    {%- endif %}
+{%- endif %}

config.json

@@ -0,0 +1,75 @@
+{
+  "architectures": [
+    "Qwen3_5ForCausalLM"
+  ],
+  "attention_bias": false,
+  "attention_dropout": 0.0,
+  "attn_output_gate": true,
+  "bos_token_id": null,
+  "dtype": "bfloat16",
+  "eos_token_id": 248044,
+  "full_attention_interval": 4,
+  "head_dim": 256,
+  "hidden_act": "silu",
+  "hidden_size": 1024,
+  "initializer_range": 0.02,
+  "intermediate_size": 3584,
+  "layer_types": [
+    "linear_attention",
+    "linear_attention",
+    "linear_attention",
+    "full_attention",
+    "linear_attention",
+    "linear_attention",
+    "linear_attention",
+    "full_attention",
+    "linear_attention",
+    "linear_attention",
+    "linear_attention",
+    "full_attention",
+    "linear_attention",
+    "linear_attention",
+    "linear_attention",
+    "full_attention",
+    "linear_attention",
+    "linear_attention",
+    "linear_attention",
+    "full_attention",
+    "linear_attention",
+    "linear_attention",
+    "linear_attention",
+    "full_attention"
+  ],
+  "linear_conv_kernel_dim": 4,
+  "linear_key_head_dim": 128,
+  "linear_num_key_heads": 16,
+  "linear_num_value_heads": 16,
+  "linear_value_head_dim": 128,
+  "mamba_ssm_dtype": "float32",
+  "max_position_embeddings": 262144,
+  "mlp_only_layers": [],
+  "model_type": "qwen3_5_text",
+  "mtp_num_hidden_layers": 1,
+  "mtp_use_dedicated_embeddings": false,
+  "num_attention_heads": 8,
+  "num_hidden_layers": 24,
+  "num_key_value_heads": 2,
+  "pad_token_id": null,
+  "partial_rotary_factor": 0.25,
+  "rms_norm_eps": 1e-06,
+  "rope_parameters": {
+    "mrope_interleaved": true,
+    "mrope_section": [
+      11,
+      11,
+      10
+    ],
+    "partial_rotary_factor": 0.25,
+    "rope_theta": 10000000,
+    "rope_type": "default"
+  },
+  "tie_word_embeddings": true,
+  "transformers_version": "5.17.0",
+  "use_cache": true,
+  "vocab_size": 248320
+}

decider/engine.py

@@ -0,0 +1,202 @@
+"""Low-latency inference engine: shape-bucketed CUDA graphs over the one-pass decision model.
+
+Right padding + causal layers => pad positions never influence earlier slots, so no attention
+mask is needed and every (B, T) bucket can be captured once and replayed.  The graph outputs
+option-letter logits for all positions [B, T, K]; slots are gathered outside.
+"""
+import time, torch, torch._dynamo, torch.nn.functional as F
+from decider.model import DecisionModel, collate
+from decider.prompt import build, MAX_OPTIONS
+
+T_BUCKETS = [64, 128, 192, 256, 320, 384, 512, 640, 768, 1024, 1280, 1536, 2048]
+B_BUCKETS = [1, 2, 4, 8, 16, 32, 64]
+GRAPH_MAX_T = 2048          # longer inputs (up to the 32k request budget) run eagerly: compute dominates there, and one graph
+LONG_STEP = 1024            # per (B, T) shape would cost a compile + capture for every new length
+
+
+def _bucket(x, buckets):
+    for b in buckets:
+        if x <= b:
+            return b
+    return None
+
+
+def fused_causal_conv1d_fn(hidden_states, weight, bias=None, activation=None, **kwargs):
+    """Depthwise causal conv (kernel k) as k shifted multiply-adds: fuses under torch.compile,
+    unlike the cuDNN grouped conv fallback (which was ~11% of batched GPU time)."""
+    B, C, T = hidden_states.shape; k = weight.shape[-1]
+    x = F.pad(hidden_states.to(weight.dtype), (k - 1, 0))
+    out = x[:, :, k - 1:k - 1 + T] * weight[:, k - 1][None, :, None]
+    for j in range(k - 1):
+        out = out + x[:, :, j:j + T] * weight[:, j][None, :, None]
+    if bias is not None:
+        out = out + bias[None, :, None]
+    if activation == "silu":
+        out = F.silu(out)
+    elif activation is not None:
+        from transformers.activations import ACT2FN
+        out = ACT2FN[activation](out)
+    return out.to(hidden_states.dtype)
+
+
+def patch_conv():
+    from transformers.models.qwen3_5 import modeling_qwen3_5 as mq
+    mq.causal_conv1d_fn = fused_causal_conv1d_fn
+
+
+def read_slots(out, rows, slots, nopts, temperature, n_per_item):
+    """One gather + one softmax + one device-to-host copy for the whole batch (was: three small kernels and a sync per item).
+    out [B, T, K] logits; rows/slots/nopts: flat python lists, one entry per question; n_per_item: questions per item."""
+    dev = out.device; idx = torch.tensor([rows, slots, nopts], dtype=torch.long).to(dev, non_blocking=True)
+    lg = out[idx[0], idx[1]]                                                                  # [N, K]
+    lg = lg.masked_fill(torch.arange(lg.shape[1], device=dev)[None, :] >= idx[2][:, None], float("-inf"))
+    p = torch.softmax(lg / temperature, -1).cpu()
+    return list(torch.split(p, n_per_item))
+
+
+def fill_ids(items_ids, B, T, pad):
+    import numpy as np
+    a = np.full((B, T), pad, dtype=np.int64)
+    for b, x in enumerate(items_ids): a[b, :len(x)] = x
+    return torch.from_numpy(a)
+
+
+class Engine:
+    """compile: torch.compile the forward (needs use_cache=False; ~1.4x batched, fuses elementwise work).
+    fp8: e4m3 weights + per-token activation scaling on the big linears (Hopper tensor cores).
+    conv_patch: fusable depthwise causal conv instead of the cuDNN fallback."""
+    def __init__(self, path, device="cuda", dtype=torch.bfloat16, use_graphs=True, max_ctx_tokens=1536,
+                 compile=True, fp8=False, conv_patch=True):
+        if conv_patch:
+            patch_conv()
+        self.m = DecisionModel(path, dtype=dtype, grad_ckpt=False).to(device).eval()
+        self.tok = self.m.tok; self.dev = device; self.use_graphs = use_graphs; self.max_ctx = max_ctx_tokens
+        self.core, self.W = self.m.lm.model, self.m.lm.lm_head.weight[self.m.letters].detach().clone()
+        self.cfg = dict(compile=compile, fp8=fp8, conv_patch=conv_patch, graphs=use_graphs)
+        if fp8:
+            from decider.fp8 import convert_to_fp8
+            self.cfg["fp8_layers"] = convert_to_fp8(self.core)
+        if compile:
+            torch._dynamo.config.cache_size_limit = 128
+            self._fwd_impl = torch.compile(self._fwd_eager, dynamic=False)
+        else:
+            self._fwd_impl = self._fwd_eager
+        self.graphs = {}                       # (B, T) -> (static_ids, static_out, graph)
+        self.pool = torch.cuda.graph_pool_handle() if use_graphs else None
+        self.stats = dict(graph_captures=0, forwards=0)
+
+    def _fwd_eager(self, ids):
+        h = self.core(input_ids=ids, use_cache=False).last_hidden_state
+        return F.linear(h, self.W).float()                          # [B, T, K]
+
+    @torch.no_grad()
+    def _fwd(self, ids):
+        return self._fwd_impl(ids)
+
+    def _capture(self, B, T):
+        s_ids = torch.full((B, T), self.tok.pad_token_id, dtype=torch.long, device=self.dev)
+        st = torch.cuda.Stream(); st.wait_stream(torch.cuda.current_stream())
+        with torch.cuda.stream(st):
+            for _ in range(3): self._fwd(s_ids)                     # warm-up: compile / triton autotune
+        torch.cuda.current_stream().wait_stream(st)
+        g = torch.cuda.CUDAGraph()
+        with torch.cuda.graph(g, pool=self.pool):
+            s_out = self._fwd(s_ids)
+        self.stats["graph_captures"] += 1
+        return s_ids, s_out, g
+
+    @torch.no_grad()
+    def logits_all(self, ids):
+        """ids: [B, T] long on device (already right-padded to a bucket). Returns [B, T, K] float."""
+        B, T = ids.shape; self.stats["forwards"] += 1
+        if T > GRAPH_MAX_T:
+            self.stats["long_forwards"] = self.stats.get("long_forwards", 0) + 1
+            return self._fwd_eager(ids)
+        if not self.use_graphs:
+            return self._fwd(ids)
+        key = (B, T)
+        if key not in self.graphs:
+            self.graphs[key] = self._capture(B, T)
+        s_ids, s_out, g = self.graphs[key]
+        s_ids.copy_(ids); g.replay()
+        return s_out
+
+    @torch.no_grad()
+    def score_items(self, items, temperature=1.0):
+        """items: list of dicts from prompt.build. Returns list of [n_q, MAX_OPTIONS] prob tensors (cpu)."""
+        Tmax = max(len(it["ids"]) for it in items)
+        T = _bucket(Tmax, T_BUCKETS) or -(-Tmax // LONG_STEP) * LONG_STEP
+        B = (_bucket(len(items), B_BUCKETS) or len(items)) if T <= GRAPH_MAX_T else len(items)
+        ids = fill_ids([it["ids"] for it in items], B, T, self.tok.pad_token_id)
+        out = self.logits_all(ids.to(self.dev, non_blocking=True))
+        return read_slots(out, [b for b, it in enumerate(items) for _ in it["slots"]], [s for it in items for s in it["slots"]],
+                          [n for it in items for n in it["nopts"]], temperature, [len(it["slots"]) for it in items])
+
+    @torch.no_grad()
+    def score_shared(self, items, temperature=1.0, min_prefix=192):
+        """Rows that start with the same tokens (one state, one question per row): run the shared prefix once, fork its
+        cache (attention KV + delta-net conv/recurrent states) to every row, and run only the question suffixes.
+        Same answers as score_items up to kernel round-off; cost ~ state + sum(questions) instead of n * state."""
+        ids = [it["ids"] for it in items]; n = len(ids)
+        lcp = 0; short = min(len(x) for x in ids) - 1
+        while lcp < short and all(x[lcp] == ids[0][lcp] for x in ids): lcp += 1
+        if n < 2 or lcp < min_prefix:
+            return self.score_items(items, temperature)
+        self.stats["shared_prefix_calls"] = self.stats.get("shared_prefix_calls", 0) + 1
+        pre = torch.tensor(ids[0][:lcp], device=self.dev)[None]
+        cache = self.core(input_ids=pre, use_cache=True).past_key_values
+        cache.reorder_cache(torch.zeros(n, dtype=torch.long, device=self.dev))            # fork: every row gets a copy of row 0
+        Ts = max(len(x) for x in ids) - lcp
+        suf = fill_ids([x[lcp:] for x in ids], n, Ts, self.tok.pad_token_id)
+        h = self.core(input_ids=suf.to(self.dev), past_key_values=cache, use_cache=True).last_hidden_state
+        rows = [b for b, it in enumerate(items) for _ in it["slots"]]; sl = [s - lcp for it in items for s in it["slots"]]
+        idx = torch.tensor([rows, sl], device=self.dev)
+        return read_slots(F.linear(h[idx[0], idx[1]], self.W).float()[:, None, :], list(range(len(rows))), [0] * len(rows),
+                          [n for it in items for n in it["nopts"]], temperature, [len(it["slots"]) for it in items])
+
+    def warmup(self, shapes=((1, 128), (1, 256), (1, 384), (1, 512), (8, 256), (8, 512), (32, 256), (32, 512))):
+        t = time.time()
+        for B, T in shapes:
+            self.logits_all(torch.full((B, T), self.tok.pad_token_id, dtype=torch.long, device=self.dev))
+        torch.cuda.synchronize(); return time.time() - t
+
+
+if __name__ == "__main__":
+    import sys, random, numpy as np
+    from decider import data as D
+    from decider.infer import Decider
+    path = sys.argv[1] if len(sys.argv) > 1 else "runs/r3_v2/model"
+    cfg = dict(compile="nocompile" not in sys.argv[2:], fp8="fp8" in sys.argv[2:], conv_patch="noconv" not in sys.argv[2:])
+    _, evals = D.load_cache("data/tasks.pkl")
+    eng = Engine(path, **cfg); print("engine cfg", eng.cfg)
+    rng = random.Random(0)
+    exs = evals["support_tickets"][:64] + evals["clinc_oos"][:64] + evals["race"][:32]
+    items = [build(e, eng.tok, rng, max_ctx_tokens=1536) for e in exs]
+    # correctness vs eager masked forward (DecisionModel.slot_logits)
+    ref = []
+    with torch.no_grad():
+        for i in range(0, len(items), 16):
+            b = collate(items[i:i + 16], eng.tok.pad_token_id)
+            lg = eng.m.slot_logits(b["input_ids"].cuda(), b["attention_mask"].cuda(), b["slot_idx"].cuda(), b["slot_batch"].cuda(), b["nopts"].cuda())
+            ref.append(torch.softmax(lg, -1).cpu())
+    ref = torch.cat(ref)
+    got = torch.cat(eng.score_items(items))
+    print(f"max |p_graph - p_eager| = {(ref - got).abs().max():.4f} over {len(ref)} questions; argmax agreement {(ref.argmax(1) == got.argmax(1)).float().mean():.4f}")
+    print(f"warmup capture of 8 buckets: {eng.warmup():.1f}s; captures so far {eng.stats['graph_captures']}")
+    # latency: single real requests
+    for name, pool in [("support_tickets", exs[:64]), ("clinc_oos", exs[64:128]), ("race", exs[128:])]:
+        its = [build(e, eng.tok, rng) for e in pool]
+        ts = []
+        for it in its[:40]:
+            torch.cuda.synchronize(); t = time.time(); eng.score_items([it]); torch.cuda.synchronize(); ts.append(time.time() - t)
+        ts = np.array(ts[5:]) * 1000
+        print(f"single request {name:16s}: p50 {np.median(ts):5.1f} ms  p90 {np.percentile(ts, 90):5.1f} ms  (avg {np.mean([len(i['ids']) for i in its]):.0f} tok, {len(its[0]['slots'])} q)")
+        for bs in (8, 32):
+            ts = []
+            for i in range(0, min(len(its), bs * 6), bs):
+                chunk = its[i:i + bs]
+                if len(chunk) < bs: break
+                torch.cuda.synchronize(); t = time.time(); eng.score_items(chunk); torch.cuda.synchronize(); ts.append(time.time() - t)
+            ts = np.array(ts[1:]) * 1000
+            print(f"   batch {bs:2d}: p50 {np.median(ts):6.1f} ms -> {bs/np.median(ts)*1000:6.0f} ctx/s, {bs*len(its[0]['slots'])/np.median(ts)*1000:6.0f} decisions/s")
+    print("stats", eng.stats, "graphs", len(eng.graphs), f"mem {torch.cuda.memory_reserved()/1e9:.1f} GB")

decider/fp8.py

@@ -0,0 +1,54 @@
+"""FP8 (e4m3) linear layers for Hopper via torch._scaled_mm.
+Weights: per-output-channel scales, quantised once.  Activations: per-token dynamic scales.
+Under torch.compile the quantisation ops fuse into the surrounding elementwise work."""
+import torch, torch.nn as nn
+
+E4M3_MAX = 448.0
+
+
+def _quant_rowwise(x):
+    s = x.abs().amax(dim=-1, keepdim=True).float().clamp(min=1e-12) / E4M3_MAX
+    return (x.float() / s).clamp(-E4M3_MAX, E4M3_MAX).to(torch.float8_e4m3fn), s
+
+
+class FP8Linear(nn.Module):
+    def __init__(self, lin: nn.Linear):
+        super().__init__()
+        wq, sw = _quant_rowwise(lin.weight.detach())          # [N,K] fp8, [N,1]
+        self.register_buffer("wq", wq.contiguous())               # [N,K]; passed as wq.t() -> [K,N] column-major, as _scaled_mm wants
+        self.register_buffer("sw_t", sw.t().contiguous())        # [1,N]
+        self.bias = None if lin.bias is None else nn.Parameter(lin.bias.detach().clone(), requires_grad=False)
+        self.in_features, self.out_features = lin.in_features, lin.out_features
+        self.out_dtype = lin.weight.dtype
+
+    def forward(self, x):
+        shp = x.shape[:-1]
+        x2 = x.reshape(-1, self.in_features)
+        xq, sx = _quant_rowwise(x2)
+        y = torch._scaled_mm(xq, self.wq.t(), scale_a=sx, scale_b=self.sw_t, bias=self.bias, out_dtype=self.out_dtype)
+        return y.reshape(*shp, self.out_features)
+
+
+def convert_to_fp8(model, skip=("lm_head",), min_dim=1024):
+    """Replace nn.Linear (with in/out >= min_dim) by FP8Linear in place. Returns count."""
+    n = 0
+    for name, mod in list(model.named_modules()):
+        for cname, child in list(mod.named_children()):
+            full = f"{name}.{cname}" if name else cname
+            if isinstance(child, nn.Linear) and not any(s in full for s in skip) and min(child.in_features, child.out_features) >= min_dim:
+                setattr(mod, cname, FP8Linear(child)); n += 1
+    return n
+
+
+if __name__ == "__main__":
+    import time
+    lin = nn.Linear(2048, 6144, bias=False).cuda().to(torch.bfloat16)
+    f8 = FP8Linear(lin)
+    x = torch.randn(8192, 2048, device="cuda", dtype=torch.bfloat16)
+    ref = lin(x); got = f8(x)
+    print("rel err", ((ref.float() - got.float()).abs().mean() / ref.float().abs().mean()).item())
+    for f, name in [(lin, "bf16 linear"), (f8, "fp8 linear (eager)"), (torch.compile(f8), "fp8 linear (compiled)")]:
+        for _ in range(3): f(x)
+        torch.cuda.synchronize(); t = time.time()
+        for _ in range(20): f(x)
+        torch.cuda.synchronize(); print(f"{name:24s} {(time.time()-t)/20*1000:.3f} ms")

decider/infer.py

@@ -0,0 +1,262 @@
+"""Usable inference API: typed decisions with probabilities, all from one forward pass.
+
+    from decider.infer import Decider
+    d = Decider("runs/r2_full/model")
+    out = d.decide("My card was charged twice for the same purchase.",
+                   [{"question": "Which department should handle this?", "options": ["billing", "technical", "sales"]},
+                    {"question": "How urgent is this?", "options": ["low", "medium", "high"]}])
+    # -> [{'choice': 'billing', 'confidence': 0.97, 'probs': {...}}, {...}]
+"""
+import torch
+from decider.model import DecisionModel, collate
+from decider.prompt import build, MAX_OPTIONS
+from dataclasses import dataclass
+
+
+@dataclass
+class Q:
+    text: str; options: list; gold: int = 0
+
+
+@dataclass
+class Example:
+    context: str; qs: list; task: str = "infer"; image: bytes = None
+
+
+NEUTRAL_NONE = "not listed here"
+
+
+def neutralize_options(options):
+    """The training augmentation used the literal 'none of the above', and the model learned that exact string as an
+    abstain signal (it abstains even on clear cases when the string is offered). Any option that reads like it is
+    rewritten to a neutral phrasing for the model and mapped back in the output."""
+    out, back = [], {}
+    for o in options:
+        key = o.strip().lower()
+        if key.startswith("none of the above") or key in ("none of the above", "none", "n/a", "none of these"):
+            out.append(NEUTRAL_NONE); back[NEUTRAL_NONE] = o
+        else:
+            out.append(o)
+    return out, back
+
+
+class CompiledSchema:
+    def __init__(self, d, rqs, h, index): self.d, self.rqs, self.h, self.index = d, rqs, h, index
+
+    def batch(self, states, max_state_tokens=32768):
+        from decider.systemone import render_state, assemble
+        probs = self.d._se.score(self.h, [render_state(s) for s in states], temperature=self.d.T_schema, max_ctx_tokens=max_state_tokens)
+        return [{"model": self.d.name, "answers": assemble(self.rqs, self.index, [p.tolist() for p in pr])} for pr in probs]
+
+    def __call__(self, state, max_state_tokens=32768):
+        return self.batch([state], max_state_tokens)[0]
+
+
+class Decider:
+    """use_graphs=True (default on CUDA) routes scoring through decider.engine.Engine: shape-bucketed
+    CUDA graphs, ~7x lower single-request latency than eager. Set False for CPU or debugging."""
+    def __init__(self, path, device="cuda", dtype=torch.bfloat16, temperature=None, abstain_below=0.0, use_graphs=None):
+        import json, os
+        cfg = {}
+        try:                                          # model folder may carry decider_config.json (temperature, flags)
+            from huggingface_hub import hf_hub_download
+            cfg_path = os.path.join(path, "decider_config.json") if os.path.isdir(path) else hf_hub_download(path, "decider_config.json")
+            cfg = json.load(open(cfg_path))
+        except Exception:
+            pass
+        if temperature is None:
+            temperature = float(cfg.get("temperature", 1.0))
+        self.neutralize_none = bool(cfg.get("neutralize_none", True))   # v4 and earlier learned the literal string as an abstain signal
+        if use_graphs is None:
+            use_graphs = str(device).startswith("cuda")
+        if use_graphs:
+            from decider.engine import Engine
+            self.eng = Engine(path, device=device, dtype=dtype); self.m = self.eng.m
+        else:
+            self.eng = None; self.m = DecisionModel(path, dtype=dtype, grad_ckpt=False).to(device).eval()
+        self.dev = device; self.T = temperature; self.abstain_below = abstain_below
+        self.name = "decider-" + str(cfg.get("version", "dev"))
+        self.schema_first = bool(cfg.get("schema_first", False)) and self.eng is not None      # default layout. Questions-first (the cacheable one) costs accuracy
+        self.T_schema = float(cfg.get("temperature_schema_first", temperature))                # (about 1.5 points on fixed label sets, more elsewhere): opt in with schema()
+        self.isolated_levels = bool(cfg.get("isolated_levels", False))      # Score levels judged one per row (v8+)
+        self._se = None; self._schemas = {}
+
+    @torch.no_grad()
+    def decide_batch(self, requests, max_ctx_tokens=1536):
+        """requests: list of (context:str, questions:list[dict(question, options)]). One forward pass for everything."""
+        exs, meta = [], []
+        if self.neutralize_none:
+            requests = [(context, [dict(q, options=neutralize_options(q["options"])[0], _back=neutralize_options(q["options"])[1]) for q in qs]) for context, qs in requests]
+        for context, qs in requests:
+            for q in qs:
+                assert 2 <= len(q["options"]) <= MAX_OPTIONS, f"2..{MAX_OPTIONS} options required"
+            exs.append(Example(context, [Q(q["question"], list(q["options"]), 0) for q in qs], "infer"))
+        class _NoShuffle:                      # keep option order as given
+            def shuffle(self, x): pass
+            def sample(self, xs, k): return xs[:k]
+        items = [build(e, self.m.tok, _NoShuffle(), max_options=MAX_OPTIONS, max_ctx_tokens=max_ctx_tokens) for e in exs]
+        if self.eng is not None:
+            probs = torch.cat(self.eng.score_items(items, temperature=self.T))
+        else:
+            b = collate(items, self.m.tok.pad_token_id)
+            logits = self.m.slot_logits(b["input_ids"].to(self.dev), b["attention_mask"].to(self.dev), b["slot_idx"].to(self.dev),
+                                        b["slot_batch"].to(self.dev), b["nopts"].to(self.dev))
+            probs = torch.softmax(logits / self.T, -1).cpu()
+        out, k = [], 0
+        for context, qs in requests:
+            res = []
+            for q in qs:
+                p = probs[k, :len(q["options"])].tolist(); k += 1
+                j = max(range(len(p)), key=p.__getitem__); back = q.get("_back", {})
+                names = [back.get(o, o) for o in q["options"]]
+                res.append(dict(choice=names[j] if p[j] >= self.abstain_below else None, confidence=p[j],
+                                probs={o: pi for o, pi in zip(names, p)}, probs_list=p))
+            out.append(res)
+        return out
+
+    def decide(self, context, questions, **kw):
+        return self.decide_batch([(context, questions)], **kw)[0]
+
+    # ---- Jev-shaped interface (decider.systemone): state + {id: Choice | Score | Noul with criteria}
+    # ---- schema cache (v7+): the questions are run once, requests only run the state (decider.schema_engine)
+    def schema(self, questions, independent=True, isolated=None, compile=False):
+        """Compile a fixed set of Jev-shaped questions: schema(state) -> answers; schema.batch([state, ...]) -> [answers]."""
+        import json
+        from decider.schema_engine import SchemaEngine
+        from decider.systemone import render_question
+        isolated = self.isolated_levels if isolated is None else isolated
+        key = (json.dumps(questions, sort_keys=True, ensure_ascii=False), independent, isolated)
+        if key not in self._schemas:
+            if self._se is None: self._se = SchemaEngine(self.eng)
+            if len(self._schemas) >= 64:                                  # drop the oldest schema and its graphs
+                old = next(iter(self._schemas)); hid = self._schemas.pop(old)[1].id
+                for k in [k for k in self._se.graphs if k[0] == hid]: del self._se.graphs[k]
+            from decider.systemone import plan_rows
+            rqs = {k: render_question(v) for k, v in questions.items()}
+            rows, index = plan_rows(rqs, isolated and independent)
+            h = self._se.prepare(rows, independent=independent, compile=compile)      # compile=True: ~25 s per (batch, length) shape, 1.6x faster after
+            self._schemas[key] = (rqs, h, index)
+        return CompiledSchema(self, *self._schemas[key])
+
+    def system_one(self, state, questions, independent=True, max_state_tokens=32768, max_fwd_tokens=65536, layout=None, isolated=None):
+        layout = layout or ("schema_first" if self.schema_first else "state_first")
+        isolated = (self.isolated_levels if isolated is None else isolated) and independent
+        if layout == "schema_first" and self.eng is not None:
+            return self.schema(questions, independent, isolated)(state, max_state_tokens)
+        """independent=True scores every question in its own row (state + that question only), so adding, removing or
+        reordering questions cannot change any other answer; the state is run once and its cache forked to every
+        question (Engine.score_shared).  independent=False packs all questions behind one copy of the state in one row
+        (later questions can then see earlier question texts)."""
+        from decider.systemone import render_state, render_question, unique_tokens, plan_rows, assemble
+        ctx = render_state(state); rqs = {k: render_question(v) for k, v in questions.items()}
+        opts = (lambda r: neutralize_options(r["options"])[0]) if self.neutralize_none else (lambda r: list(r["options"]))
+        flat, index = plan_rows(rqs, isolated)
+        rows = [[r] for r in flat] if independent else [flat]
+        class _Keep:
+            def shuffle(self, x): pass
+            def sample(self, xs, k): return xs[:k]
+        items = [build(Example(ctx, [Q(r["question"], opts(r), 0) for r in row]), self.m.tok, _Keep(), max_options=MAX_OPTIONS,
+                       max_ctx_tokens=max_state_tokens, layout=layout) for row in rows]
+        with torch.no_grad():
+            if self.eng is not None and len(items) > 1 and layout == "state_first":
+                probs = self.eng.score_shared(items, temperature=self.T)
+            else:
+                probs = []; per = max(1, max_fwd_tokens // max(len(it["ids"]) for it in items))
+                for i in range(0, len(items), per):
+                    if self.eng is not None:
+                        probs += self.eng.score_items(items[i:i + per], temperature=self.T)
+                    else:
+                        bt = collate(items[i:i + per], self.m.tok.pad_token_id)
+                        lg = self.m.slot_logits(*[bt[k].to(self.dev) for k in ("input_ids", "attention_mask", "slot_idx", "slot_batch", "nopts")])
+                        pr = torch.softmax(lg / self.T, -1).cpu(); c = 0
+                        for it in items[i:i + per]:
+                            probs.append(pr[c:c + len(it["slots"])]); c += len(it["slots"])
+        flatp = [p.tolist() for ps in probs for p in ps]
+        return {"model": self.name, "answers": assemble(rqs, index, flatp),
+                "usage": {"input_tokens": unique_tokens(items), "output_tokens": 0}}
+
+    # ---- typed schema interface: {question: {"type": "bool"} | {"type": "choice", "options": [...]}
+    #                                         | {"type": "scale", "legend": {"0": "none", "1": "low", ...}}}
+    @staticmethod
+    def _schema_to_questions(schema):
+        qs = []
+        for qtext, spec in schema.items():
+            t = spec.get("type", "choice")
+            if t == "bool":
+                qs.append(dict(question=qtext, options=["no", "yes"]))
+            elif t == "choice":
+                qs.append(dict(question=qtext, options=list(spec["options"])))
+            elif t == "scale":
+                leg = spec["legend"]
+                keys = sorted(leg, key=lambda k: float(k)) if isinstance(leg, dict) else list(range(len(leg)))
+                labels = [f"{k}: {leg[k]}" if isinstance(leg, dict) else f"{i}: {leg[i]}" for i, k in enumerate(keys)]
+                qs.append(dict(question=qtext, options=labels, _keys=keys, _legend=leg))
+            else:
+                raise ValueError(f"unknown field type {t}")
+        return qs
+
+    def decide_json_batch(self, requests, **kw):
+        """requests: list of (context, schema). Returns one dict per context keyed by question."""
+        qss = [self._schema_to_questions(schema) for _, schema in requests]
+        raw = self.decide_batch([(ctx, qs) for (ctx, _), qs in zip(requests, qss)], **kw)
+        out = []
+        for (ctx, schema), qs, res in zip(requests, qss, raw):
+            o = {}
+            for (qtext, spec), q, r in zip(schema.items(), qs, res):
+                t = spec.get("type", "choice")
+                if t == "bool":
+                    o[qtext] = {"noul": round(r["probs"]["yes"], 4), "type": "noul"}
+                elif t == "choice":
+                    o[qtext] = {"choice": r["choice"], "confidence": round(r["confidence"], 4), "type": "choice",
+                                "probabilities": {k: round(v, 4) for k, v in r["probs"].items()}}
+                else:
+                    p = [r["probs"][lab] for lab in q["options"]]
+                    keys = q["_keys"]; n = len(p)
+                    score = sum(float(k) * pi for k, pi in zip(keys, p))          # expected level on the legend scale
+                    j = max(range(n), key=p.__getitem__)
+                    o[qtext] = {"score": round(score, 2), "confidence": round(p[j], 4), "type": "scale", "legend": q["_legend"],
+                                "probabilities": {str(keys[i]): round(pi, 4) for i, pi in enumerate(p)}}
+            out.append(o)
+        return out
+
+    def decide_json(self, context, schema, **kw):
+        return self.decide_json_batch([(context, schema)], **kw)[0]
+
+
+if __name__ == "__main__":
+    import sys, json, time
+    d = Decider(sys.argv[1] if len(sys.argv) > 1 else "runs/r1_200k/model")
+    demo = [
+        ("My card was charged twice for the same purchase and I want the extra charge refunded.",
+         [{"question": "Which department should handle this?", "options": ["billing", "technical support", "sales"]},
+          {"question": "What is the customer's sentiment?", "options": ["angry", "neutral", "happy"]},
+          {"question": "Does this need a refund action?", "options": ["no", "yes"]}]),
+        ("hey can u turn the lights off in the kitchen",
+         [{"question": "What is the intent?", "options": ["smart home control", "set alarm", "play music", "none of the above"]},
+          {"question": "Is this request toxic?", "options": ["no", "yes"]}]),
+        ("The quarterly report shows revenue fell 12% while costs rose sharply.",
+         [{"question": "What is the financial sentiment?", "options": ["bearish", "neutral", "bullish"]}]),
+    ]
+    t = time.time(); res = d.decide_batch(demo); dt = time.time() - t
+    for (ctx, qs), r in zip(demo, res):
+        print("\n>>", ctx)
+        for q, a in zip(qs, r):
+            print(f"   {q['question']:45s} -> {a['choice']!s:22s} p={a['confidence']:.2f}  " + " ".join(f"{o}:{p:.2f}" for o, p in a['probs'].items()))
+    print(f"\n{sum(len(q) for _, q in demo)} decisions in {dt*1000:.0f} ms (one forward pass)")
+    schema = {
+        "Revenue currently impacted?": {"type": "bool"},
+        "What business impact?": {"type": "choice", "options": ["none", "degraded", "outage"]},
+        "Integration issue present?": {"type": "bool"},
+        "Account health status?": {"type": "choice", "options": ["healthy", "watch", "at risk"]},
+        "Which incident scope?": {"type": "choice", "options": ["single_account", "multi_account", "platform_wide"]},
+        "Security concern present?": {"type": "bool"},
+        "Duplicate charge reported?": {"type": "bool"},
+        "Churn likelihood level?": {"type": "scale", "legend": {"0": "none", "1": "low", "2": "medium", "3": "high"}},
+        "Human attention needed?": {"type": "bool"},
+        "Immediate feature request?": {"type": "bool"},
+    }
+    ctx = ("Hi, since this morning our Stripe webhook integration stopped firing and our checkout is down for all customers. "
+           "We are losing orders every minute and our partner launch is on Thursday. Also I think we got billed twice last week. "
+           "If this is not fixed today we will have to look at other providers.")
+    t = time.time(); js = d.decide_json(ctx, schema); dt = time.time() - t
+    print(f"\n>> {ctx[:80]}...\n" + json.dumps(js, indent=1)[:3000]); print(f"{len(schema)} typed fields in {dt*1000:.0f} ms (one forward pass)")

decider/metrics.py

@@ -0,0 +1,37 @@
+import numpy as np
+
+
+def ece(conf, correct, bins=15):
+    conf = np.asarray(conf); correct = np.asarray(correct, dtype=float)
+    edges = np.linspace(0, 1, bins + 1); e = 0.0
+    for lo, hi in zip(edges[:-1], edges[1:]):
+        m = (conf > lo) & (conf <= hi)
+        if m.any():
+            e += m.mean() * abs(conf[m].mean() - correct[m].mean())
+    return float(e)
+
+
+def aurc(conf, correct):
+    """Area under risk-coverage curve (lower is better)."""
+    order = np.argsort(-np.asarray(conf)); c = np.asarray(correct, dtype=float)[order]
+    risk = np.cumsum(1 - c) / np.arange(1, len(c) + 1)
+    return float(risk.mean())
+
+
+def sel_acc(conf, correct, coverage):
+    order = np.argsort(-np.asarray(conf)); c = np.asarray(correct, dtype=float)[order]
+    n = max(1, int(round(coverage * len(c))))
+    return float(c[:n].mean())
+
+
+def summarize(probs, golds, nopts):
+    """probs [N,K] (masked entries 0), golds [N], nopts [N]."""
+    probs = np.asarray(probs); golds = np.asarray(golds); nopts = np.asarray(nopts)
+    pred = probs.argmax(1); conf = probs.max(1); correct = (pred == golds)
+    p_gold = probs[np.arange(len(golds)), golds]
+    nll = -np.log(np.clip(p_gold, 1e-12, 1)).mean()
+    onehot = np.zeros_like(probs); onehot[np.arange(len(golds)), golds] = 1
+    brier = ((probs - onehot) ** 2).sum(1).mean()
+    return dict(n=int(len(golds)), acc=float(correct.mean()), nll=float(nll), brier=float(brier), ece=ece(conf, correct),
+                aurc=aurc(conf, correct), acc_at_80=sel_acc(conf, correct, 0.8), acc_at_50=sel_acc(conf, correct, 0.5),
+                chance=float((1.0 / nopts).mean()), mean_conf=float(conf.mean()))

decider/model.py

@@ -0,0 +1,47 @@
+"""Backbone -> slot hidden states -> restricted logits over option letters."""
+import torch, torch.nn as nn, torch.nn.functional as F
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from decider.prompt import letter_ids, MAX_OPTIONS
+
+
+class DecisionModel(nn.Module):
+    def __init__(self, name, dtype=torch.bfloat16, grad_ckpt=True):
+        super().__init__()
+        self.tok = AutoTokenizer.from_pretrained(name)
+        self.lm = AutoModelForCausalLM.from_pretrained(name, dtype=dtype)
+        if grad_ckpt:
+            self.lm.gradient_checkpointing_enable()
+        self.register_buffer("letters", torch.tensor(letter_ids(self.tok)), persistent=False)
+
+    def slot_logits(self, input_ids, attention_mask, slot_idx, slot_batch, nopts):
+        """input_ids [B,T]; slot_idx/slot_batch [N] flat slot positions; nopts [N].
+        Returns [N, MAX_OPTIONS] logits with invalid options masked to -inf."""
+        h = self.lm.model(input_ids=input_ids, attention_mask=attention_mask).last_hidden_state
+        hs = h[slot_batch, slot_idx]                                   # [N,H]
+        W = self.lm.lm_head.weight[self.letters]                       # [K,H]
+        logits = F.linear(hs, W).float()                               # [N,K]
+        ar = torch.arange(MAX_OPTIONS, device=logits.device)[None, :]
+        logits = logits.masked_fill(ar >= nopts[:, None], float("-inf"))
+        return logits
+
+    def forward(self, batch):
+        return self.slot_logits(batch["input_ids"], batch["attention_mask"], batch["slot_idx"], batch["slot_batch"], batch["nopts"])
+
+
+def collate(items, pad_id):
+    """items: list of dicts from prompt.build (+ 'task', 'ex_id'). Right-pad."""
+    T = max(len(it["ids"]) for it in items)
+    T = ((T + 63) // 64) * 64          # few distinct shapes -> fewer kernel (re)compiles
+    B = len(items)
+    input_ids = torch.full((B, T), pad_id, dtype=torch.long)
+    attn = torch.zeros((B, T), dtype=torch.long)
+    slot_idx, slot_batch, golds, nopts, tasks, qidx = [], [], [], [], [], []
+    for b, it in enumerate(items):
+        n = len(it["ids"])
+        input_ids[b, :n] = torch.tensor(it["ids"])
+        attn[b, :n] = 1
+        for k, s in enumerate(it["slots"]):
+            slot_idx.append(s); slot_batch.append(b); golds.append(it["golds"][k]); nopts.append(it["nopts"][k])
+            tasks.append(it.get("task", "")); qidx.append(k)
+    return dict(input_ids=input_ids, attention_mask=attn, slot_idx=torch.tensor(slot_idx), slot_batch=torch.tensor(slot_batch),
+                golds=torch.tensor(golds), nopts=torch.tensor(nopts), tasks=tasks, qidx=qidx)

decider/prompt.py

@@ -0,0 +1,157 @@
+"""Prompt construction.  One context, N typed questions, N answer slots.
+
+All N decisions are read from a single forward pass: the logits at each
+"Answer k: (" slot are restricted to the option-letter tokens.  No answer
+letters are ever inserted, so slot k sees the context and all questions but
+no earlier answers (the decisions are conditionally independent given input).
+"""
+import random
+
+LETTERS = "ABCDEFGHIJ"
+NARROW = len(LETTERS)      # <= NARROW options: the original "(A) .. (J)" rendering, tokenized as a string (unchanged since v1)
+MAX_OPTIONS = 255          # width of the label head. > NARROW options: "wide" rendering, one label token per option:
+                           # A..Z then the first 229 two-letter upper-case strings that are single tokens (AA, AB, ...)
+ABSTAIN_PREFIXES = ("none of the above", "none of these", "not listed", "no suitable", "does not apply", "cannot tell")
+ABSTAIN_EXACT = ("other", "unsure", "something else", "neither of these", "other / not covered")
+
+
+def is_abstain_option(o):
+    o = o.strip().lower()
+    return o.startswith(ABSTAIN_PREFIXES) or o in ABSTAIN_EXACT
+
+
+_LABELS = {}
+_OPT_CACHE = {}
+
+
+def _enc_opt(tok, text):
+    """Token ids of ") <option text>" (cached: fixed label sets repeat the same strings millions of times)."""
+    key = (id(tok), text)
+    v = _OPT_CACHE.get(key)
+    if v is None:
+        v = tok.encode(f") {text}", add_special_tokens=False)
+        if len(_OPT_CACHE) < 2_000_000:
+            _OPT_CACHE[key] = v
+    return v
+
+
+def label_table(tok):
+    """(label strings, label token ids), MAX_OPTIONS entries; the first NARROW are A..J so narrow questions are unchanged."""
+    key = id(tok)
+    if key not in _LABELS:
+        import string
+        U = string.ascii_uppercase
+        names = list(U) + [a + b for a in U for b in U]
+        out = []
+        for n in names:
+            t = tok.encode(n, add_special_tokens=False)
+            if len(t) == 1:
+                out.append((n, t[0]))
+            if len(out) == MAX_OPTIONS:
+                break
+        assert len(out) == MAX_OPTIONS and len({i for _, i in out}) == MAX_OPTIONS
+        _LABELS[key] = ([n for n, _ in out], [i for _, i in out],
+                        tok.encode("\n(", add_special_tokens=False))
+    return _LABELS[key]
+
+
+def _select(q, rng, max_options):
+    opts = list(range(len(q.options)))
+    if len(opts) > max_options:
+        # always keep the gold and any abstain-style option (its mere presence must not carry information)
+        forced = {q.gold} | {i for i, o in enumerate(q.options) if is_abstain_option(o)}
+        others = [i for i in opts if i not in forced]
+        opts = rng.sample(others, max_options - len(forced)) + list(forced)
+    rng.shuffle(opts)
+    return opts
+
+
+def _options_ids(tok, q, opts):
+    if len(opts) <= NARROW:
+        return tok.encode("".join(f"\n({LETTERS[j]}) {q.options[oi]}" for j, oi in enumerate(opts)), add_special_tokens=False)
+    _, lab_ids, open_ids = label_table(tok); out = []
+    for j, oi in enumerate(opts):
+        out += open_ids + [lab_ids[j]] + _enc_opt(tok, q.options[oi])
+    return out
+
+
+def build_schema_first(example, tok, rng=None, max_options=NARROW, max_ctx_tokens=1536):
+    """Schema-first layout: all question/option blocks, then the context, then one answer slot per question.
+
+        Question 1: ...\nOptions:\n(A) ...          <- prefix: depends only on the questions, so its cache (attention KV and
+        \n\nQuestion 2: ...                            delta-net states) is computed once per schema and reused for every state
+        \n\nContext:\n<state>\n\nAnswer 1: (\nAnswer 2: (
+
+    The three parts are tokenized separately, so `ids[:prefix_len]` is identical for every state."""
+    rng = rng or random
+    perms = [_select(q, rng, max_options) for q in example.qs]
+    pre = schema_prefix_ids(tok, example.qs, perms)
+    suf, slots = schema_suffix_ids(tok, example.context, len(example.qs), max_ctx_tokens)
+    return dict(ids=pre + suf, slots=[len(pre) + s for s in slots], golds=[p.index(q.gold) if q.gold in p else -1 for p, q in zip(perms, example.qs)],
+                nopts=[len(p) for p in perms], perms=perms, prefix_len=len(pre))
+
+
+def schema_prefix_ids(tok, qs, perms=None):
+    """Token ids of the question/option blocks (the cacheable part of the schema-first layout)."""
+    multi = len(qs) > 1; pre = []
+    for k, q in enumerate(qs):
+        opts = perms[k] if perms is not None else list(range(len(q.options)))
+        pre += tok.encode(f"{chr(10) * 2 if k else ''}Question{' ' + str(k + 1) if multi else ''}: {q.text}\nOptions:", add_special_tokens=False) + _options_ids(tok, q, opts)
+    return pre
+
+
+def schema_suffix_ids(tok, context, n_q, max_ctx_tokens=1536):
+    """Token ids after the schema prefix: the context and one answer slot per question. Returns (ids, slot positions in ids)."""
+    ids = tok.encode("\n\nContext:\n", add_special_tokens=False) + tok.encode(context, add_special_tokens=False)[:max_ctx_tokens]; slots = []
+    for k in range(n_q):
+        ids += tok.encode(f"{chr(10) * 2 if k == 0 else chr(10)}Answer{' ' + str(k + 1) if n_q > 1 else ''}: (", add_special_tokens=False); slots.append(len(ids) - 1)
+    return ids, slots
+
+
+def build(example, tok, rng=None, max_options=NARROW, max_ctx_tokens=1536, layout="state_first"):
+    """Returns dict(ids=list[int], slots=list[int], golds=list[int], nopts=list[int], perms=list[list[int]])."""
+    if layout == "schema_first":
+        return build_schema_first(example, tok, rng, max_options, max_ctx_tokens)
+    rng = rng or random
+    ctx_ids = tok.encode("Context:\n" + example.context, add_special_tokens=False)[:max_ctx_tokens]
+    ids = list(ctx_ids)
+    slots, golds, nopts, perms = [], [], [], []
+    multi = len(example.qs) > 1
+    for k, q in enumerate(example.qs):
+        opts = list(range(len(q.options)))
+        if len(opts) > max_options:
+            # always keep the gold and any abstain-style option (its mere presence must not carry information)
+            forced = {q.gold} | {i for i, o in enumerate(q.options) if is_abstain_option(o)}
+            others = [i for i in opts if i not in forced]
+            keep = rng.sample(others, max_options - len(forced)) + list(forced)
+            opts = keep
+        rng.shuffle(opts)
+        head = f"\n\nQuestion{' ' + str(k + 1) if multi else ''}: {q.text}\nOptions:"
+        tail = f"\nAnswer{' ' + str(k + 1) if multi else ''}: ("
+        if len(opts) <= NARROW:
+            lines = [head] + [f"\n({LETTERS[j]}) {q.options[oi]}" for j, oi in enumerate(opts)] + [tail]
+            piece = tok.encode("".join(lines), add_special_tokens=False)
+        else:                                   # wide: "\n(" + <label token> + ") text", built from ids so every label is one token
+            _, lab_ids, open_ids = label_table(tok)
+            piece = tok.encode(head, add_special_tokens=False)
+            for j, oi in enumerate(opts):
+                piece += open_ids + [lab_ids[j]] + _enc_opt(tok, q.options[oi])
+            piece += tok.encode(tail, add_special_tokens=False)
+        ids.extend(piece)
+        slots.append(len(ids) - 1)          # position of " (" token
+        golds.append(opts.index(q.gold) if q.gold in opts else -1)
+        nopts.append(len(opts))
+        perms.append(opts)
+    return dict(ids=ids, slots=slots, golds=golds, nopts=nopts, perms=perms)
+
+
+def letter_ids(tok):
+    ids = label_table(tok)[1]
+    for j, L in enumerate(LETTERS):
+        assert tok.encode(L, add_special_tokens=False) == [ids[j]], L
+    return ids
+
+
+def render(example, tok, **kw):
+    b = build(example, tok, **kw)
+    return tok.decode(b["ids"])

decider/schema_engine.py

@@ -0,0 +1,143 @@
+"""Schema cache: compute a question schema once, then score states against it.
+
+In production the questions are fixed and only the state changes.  With the schema-first prompt layout
+(prompt.build_schema_first) the question/option blocks are a prefix that does not depend on the state, so their
+cache - attention K/V for the 6 full-attention layers, conv + recurrent state for the 18 delta-net layers - is computed
+once (`prepare`).  A request then runs only "Context: <state>" plus one answer slot per question, as a CUDA graph per
+(batch, length) bucket.  The prefix cache is read-only during a request (nothing is written back), so one copy serves
+every batch and every graph.
+
+    se = SchemaEngine(engine); h = se.prepare([{"question": ..., "options": [...]}, ...])
+    probs = se.score(h, ["state 1", "state 2", ...])          # list of [n_questions, MAX_OPTIONS] tensors
+"""
+import time, types, torch, torch.nn.functional as F
+from decider.prompt import schema_prefix_ids, schema_suffix_ids, MAX_OPTIONS
+from decider.engine import read_slots, fill_ids
+
+TS_BUCKETS = [32, 48, 64, 96, 128, 192, 256, 384, 512, 768, 1024]
+B_BUCKETS = [1, 2, 4, 8, 16, 32, 64]
+
+
+class _Q:
+    def __init__(self, text, options): self.text, self.options = text, options
+
+
+class PrefixCache:
+    """Duck-typed transformers Cache over fixed, read-only prefixes, for one suffix forward pass.
+    A handle holds P prefixes (P = 1: all questions packed in one prefix; P = n_questions: one prefix per question, so every
+    question is scored independently).  A batch of R states has R * P rows; row r * P + p continues prefix p."""
+    def __init__(self, h, R):
+        rep = (lambda t: t.expand(R, *t.shape[1:])) if h.P == 1 else (lambda t: t.repeat(R, *([1] * (t.dim() - 1))))
+        self.tp = h.tpmax; self.k = {i: rep(k) for i, k in h.k.items()}; self.v = {i: rep(v) for i, v in h.v.items()}
+        self.conv = {i: rep(c).contiguous() for i, c in h.conv.items()}
+        self.layers = {i: types.SimpleNamespace(record_past=False, recurrent_states={0: rep(r).contiguous()}) for i, r in h.rec.items()}
+
+    def has_previous_state(self, layer_idx=None, state_idx=None): return True
+    def get_seq_length(self, *a, **k): return self.tp
+    def update(self, key, value, layer_idx, *a, **k): return torch.cat([self.k[layer_idx], key], 2), torch.cat([self.v[layer_idx], value], 2)
+    def update_conv_state(self, x, layer_idx, **k): return torch.cat([self.conv[layer_idx].to(x.dtype), x], -1)
+    def update_recurrent_state(self, s, layer_idx, **k): return s
+
+
+class SchemaEngine:
+    def __init__(self, engine, use_graphs=True):
+        self.e = engine; self.core = engine.core; self.W = engine.W; self.tok = engine.tok; self.dev = engine.dev
+        self.use_graphs = use_graphs and engine.use_graphs; self.graphs = {}; self.stats = dict(prepared=0, captures=0, replays=0, eager=0)
+        self.compile = bool(engine.cfg.get("compile")); self._compiled = {}
+        if self.compile:                        # every compiled schema graph specialises the model frames again (its cache tensors are constants)
+            import torch._dynamo
+            torch._dynamo.config.cache_size_limit = 4096; torch._dynamo.config.accumulated_cache_size_limit = 1 << 16
+
+    @torch.no_grad()
+    def prepare(self, questions, independent=False, compile=False):
+        """questions: [{"question": str, "options": [str]}] in the order answers are wanted.  Runs the prefix(es) once.
+        independent=False: one prefix holding every question (cheapest: a request costs state + n slots).
+        independent=True:  one prefix per question, one row per question (a request costs n * (state + 1 slot); no question
+                           can influence another)."""
+        qs = [_Q(q["question"], list(q["options"])) for q in questions]
+        groups = [[q] for q in qs] if independent else [qs]; pres = [schema_prefix_ids(self.tok, g) for g in groups]
+        h = types.SimpleNamespace(P=len(groups), nq=len(qs), slots_per_row=1 if independent else len(qs), nopts=[len(q.options) for q in qs], tps=[len(p) for p in pres],
+                                  tpmax=max(len(p) for p in pres), k={}, v={}, conv={}, rec={}, id=self.stats["prepared"],
+                                  compile=bool(compile and self.compile))
+        parts = []
+        for pre in pres:
+            out = self.core(input_ids=torch.tensor(pre, device=self.dev)[None], use_cache=True).past_key_values; d = dict(k={}, v={}, conv={}, rec={})
+            for i, layer in enumerate(out.layers):
+                if getattr(layer, "recurrent_states", None) is not None and layer.recurrent_states.get(0) is not None:
+                    d["conv"][i] = layer.conv_states[0]; d["rec"][i] = layer.recurrent_states[0]
+                else:                                                   # right-pad every prefix's K/V to the longest; the mask hides the padding
+                    pad = (0, 0, 0, h.tpmax - len(pre)); d["k"][i] = F.pad(layer.keys, pad); d["v"][i] = F.pad(layer.values, pad)
+            parts.append(d)
+        for name in ("k", "v", "conv", "rec"):
+            getattr(h, name).update({i: torch.cat([d[name][i] for d in parts], 0).clone() for i in parts[0][name]})
+        self.stats["prepared"] += 1
+        return h
+
+    def _fwd(self, ids, cache, mask, pos):
+        hs = self.core(input_ids=ids, past_key_values=cache, attention_mask={"full_attention": mask, "linear_attention": None}, position_ids=pos, use_cache=True).last_hidden_state
+        return F.linear(hs, self.W).float()
+
+    def _static(self, h, R, Ts):
+        """R request slots -> R * P rows.  Mask: a row sees its own prefix (not the padding up to tpmax) and the causal suffix."""
+        ar = torch.arange(Ts, device=self.dev); tps = torch.tensor(h.tps, device=self.dev).repeat(R)                                # [R*P]
+        pre = (torch.arange(h.tpmax, device=self.dev)[None, :] < tps[:, None])[:, None, None, :].expand(-1, 1, Ts, -1)             # [B,1,Ts,tpmax]
+        mask = torch.cat([pre, (ar[:, None] >= ar[None, :])[None, None].expand(len(tps), 1, -1, -1)], 3).contiguous()
+        return PrefixCache(h, R), mask, (tps[:, None] + ar[None, :]).contiguous()
+
+    def _capture(self, h, R, Ts):
+        B = R * h.P
+        ids = torch.full((B, Ts), self.tok.pad_token_id, dtype=torch.long, device=self.dev); cache, mask, pos = self._static(h, R, Ts)
+        fwd = self._fwd
+        if h.compile:                                       # one compiled function per graph (20-30 s each: only for preloaded schemas): the cache tensors are constants of that graph
+            fwd = torch.compile(lambda i: self._fwd(i, cache, mask, pos), dynamic=False)
+            call = lambda: fwd(ids)
+        else:
+            call = lambda: fwd(ids, cache, mask, pos)
+        st = torch.cuda.Stream(); st.wait_stream(torch.cuda.current_stream())
+        with torch.cuda.stream(st):
+            for _ in range(3): call()
+        torch.cuda.current_stream().wait_stream(st)
+        g = torch.cuda.CUDAGraph()
+        with torch.cuda.graph(g, pool=self.e.pool):
+            out = call()
+        self.stats["captures"] += 1
+        return ids, out, g, (cache, mask, pos)
+
+    def warmup(self, h, batch_sizes=(1, 8, 32), state_tokens=(64, 128, 256)):
+        """Capture (and, for a compiled schema, compile) the graphs for these request-batch sizes and suffix lengths ahead of traffic."""
+        t = time.time()
+        for R in batch_sizes:
+            for Ts in state_tokens:
+                Ts = next((x for x in TS_BUCKETS if x >= Ts), TS_BUCKETS[-1])
+                if (h.id, R, Ts) not in self.graphs: self.graphs[(h.id, R, Ts)] = self._capture(h, R, Ts)
+        torch.cuda.synchronize(); return time.time() - t
+
+    def tokenize(self, h, context, max_ctx_tokens=1536):
+        """CPU part of a request (do it outside any GPU lock): -> (suffix ids, slot positions)."""
+        return schema_suffix_ids(self.tok, context, h.slots_per_row, max_ctx_tokens)
+
+    @staticmethod
+    def bucket(n_tokens):
+        return next((t for t in TS_BUCKETS if t >= n_tokens), -(-n_tokens // 256) * 256)
+
+    def score(self, h, contexts, temperature=1.0, max_ctx_tokens=1536):
+        """-> one [n_questions, MAX_OPTIONS] probability tensor per context."""
+        return self.score_rows(h, [self.tokenize(h, c, max_ctx_tokens) for c in contexts], temperature)
+
+    @torch.no_grad()
+    def score_rows(self, h, rows, temperature=1.0):
+        """rows: [(suffix ids, slots)] from tokenize()."""
+        Tmax = max(len(r[0]) for r in rows); Ts = next((t for t in TS_BUCKETS if t >= Tmax), None); n = len(rows)
+        R = next((b for b in B_BUCKETS if b >= n), n) if Ts else n; Ts = Ts or -(-Tmax // 256) * 256
+        ids = fill_ids([x for x, _ in rows for _ in range(h.P)], R * h.P, Ts, self.tok.pad_token_id).to(self.dev, non_blocking=True)
+        if self.use_graphs and Ts <= TS_BUCKETS[-1]:
+            key = (h.id, R, Ts)
+            if key not in self.graphs: self.graphs[key] = self._capture(h, R, Ts)
+            s_ids, s_out, g, _ = self.graphs[key]; s_ids.copy_(ids); g.replay(); out = s_out; self.stats["replays"] += 1
+        else:
+            out = self._fwd(ids, *self._static(h, R, Ts)); self.stats["eager"] += 1
+        if h.P == 1:                                           # packed: n slots in one row per request
+            rws = [r for r in range(n) for _ in range(h.nq)]; sls = [x for _, sl in rows for x in sl]
+        else:                                                  # independent: one slot in each of the request's P rows
+            rws = [r * h.P + p for r in range(n) for p in range(h.P)]; sls = [sl[0] for _, sl in rows for _ in range(h.P)]
+        return read_slots(out, rws, sls, h.nopts * n, temperature, [h.nq] * n)

decider/serve.py

@@ -0,0 +1,289 @@
+"""Micro-batching HTTP server.
+  POST /decide        {"context": str, "schema": {...}}                     -> typed JSON decisions (all questions packed in one row)
+  POST /v1/systemone  {"state": str|object|array, "questions": {id: {...}}} -> the TypeSafe/Jev wire format (decider.systemone):
+                      Choice (up to 255 described options), Score, Noul; every question is scored in its own row, so answers
+                      are independent of each other ("independent": false packs them behind one copy of the state instead).
+Requests arriving within `max_wait_ms` are scored in one forward pass (grouped by length bucket).
+   uvicorn decider.serve:app --host 0.0.0.0 --port 8000     (env: DECIDER_MODEL, DECIDER_MAX_BATCH, DECIDER_MAX_WAIT_MS)
+"""
+import asyncio, os, random, time, threading
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from decider.engine import Engine, T_BUCKETS, _bucket
+from decider.prompt import build, MAX_OPTIONS
+from decider.infer import Decider, Example, Q, neutralize_options
+from decider import systemone as S1
+
+MODEL = os.environ.get("DECIDER_MODEL", "runs/r3_v2/model")
+MAX_BATCH = int(os.environ.get("DECIDER_MAX_BATCH", "32"))
+MAX_WAIT_MS = float(os.environ.get("DECIDER_MAX_WAIT_MS", "8"))
+BATCH_WAIT_MS = float(os.environ.get("DECIDER_BATCH_WAIT_MS", "0"))
+MAX_STATE_TOKENS = int(os.environ.get("DECIDER_MAX_STATE_TOKENS", "32768"))
+MAX_FWD_TOKENS = int(os.environ.get("DECIDER_MAX_FWD_TOKENS", "65536"))     # padded tokens per forward pass
+COMPILE = os.environ.get("DECIDER_COMPILE", "1") == "1"
+FP8 = os.environ.get("DECIDER_FP8", "1") == "1"
+app = FastAPI(title="decider")
+MODEL_NAME = "decider"; TEMP = 1.0; TEMP_SCHEMA = 1.0; RELEASE_DATE = "2026-09-17"
+gpu_lock = threading.Lock()            # one GPU job at a time: batched graph replays and shared-prefix requests must not interleave
+SHARED_MIN_TOKENS = int(os.environ.get("DECIDER_SHARED_MIN_TOKENS", "768"))   # independent rows over a state this long share one prefix pass
+eng = None; queue = None; stats = dict(requests=0, batches=0, decisions=0, batch_hist={})
+
+
+class Req(BaseModel):
+    context: str
+    schema_: dict = None
+    model_config = {"populate_by_name": True}
+    def __init__(self, **kw):
+        if "schema" in kw: kw["schema_"] = kw.pop("schema")
+        super().__init__(**kw)
+
+
+class _NoShuffle:
+    def shuffle(self, x): pass
+    def sample(self, xs, k): return xs[:k]
+
+
+def _prepare(context, schema):
+    qs = Decider._schema_to_questions(schema)
+    for q in qs:
+        if getattr(eng, "neutralize_none", True):
+            q["options"], q["_back"] = neutralize_options(q["options"])
+    ex = Example(context, [Q(q["question"], list(q["options"]), 0) for q in qs])
+    it = build(ex, eng.tok, _NoShuffle(), max_options=MAX_OPTIONS, max_ctx_tokens=eng.max_ctx)
+    return qs, it
+
+
+def _format(schema, qs, probs):
+    o = {}
+    for (qtext, spec), q, p in zip(schema.items(), qs, probs):
+        p = p[:len(q["options"])].tolist(); t = spec.get("type", "choice"); j = max(range(len(p)), key=p.__getitem__)
+        back = q.get("_back", {}); names = [back.get(x, x) for x in q["options"]]
+        if t == "bool":
+            o[qtext] = {"noul": round(p[1], 4), "type": "noul"}
+        elif t == "choice":
+            o[qtext] = {"choice": names[j], "confidence": round(p[j], 4), "type": "choice",
+                        "probabilities": {k: round(v, 4) for k, v in zip(names, p)}}
+        else:
+            keys = q["_keys"]; score = sum(float(k) * pi for k, pi in zip(keys, p))
+            o[qtext] = {"score": round(score, 2), "confidence": round(p[j], 4), "type": "scale", "legend": q["_legend"],
+                        "probabilities": {str(keys[i]): round(pi, 4) for i, pi in enumerate(p)}}
+    return o
+
+
+async def _collect(q):
+    """Continuous batching: take what is already queued and go.  While a forward pass runs, new requests pile up and form the
+    next batch, so there is no fixed wait at low load (it cost 1.5 ms per request) and full batches at high load.
+    DECIDER_BATCH_WAIT_MS > 0 restores a short collection window."""
+    batch = [await q.get()]; deadline = time.monotonic() + BATCH_WAIT_MS / 1000
+    while len(batch) < MAX_BATCH:
+        try:
+            batch.append(q.get_nowait())
+        except asyncio.QueueEmpty:
+            timeout = deadline - time.monotonic()
+            if timeout <= 0: break
+            try: batch.append(await asyncio.wait_for(q.get(), timeout))
+            except asyncio.TimeoutError: break
+    return batch
+
+
+async def batcher():
+    loop = asyncio.get_running_loop()
+    while True:
+        batch = await _collect(queue)
+        # sort by length; split into at most two groups when the spread is large (keeps padding small)
+        batch.sort(key=lambda x: len(x[2]["ids"]))
+        groups = [batch]
+        if len(batch) >= 4:
+            lo, hi = len(batch[0][2]["ids"]), len(batch[-1][2]["ids"])
+            if _bucket(hi, T_BUCKETS) != _bucket(lo, T_BUCKETS) and hi > 1.5 * lo:
+                cut = len(batch) // 2; groups = [batch[:cut], batch[cut:]]
+        capped = []                                       # long rows: keep every forward under MAX_FWD_TOKENS padded tokens
+        for g in groups:
+            cur = []
+            for x in g:
+                if cur and (len(cur) + 1) * len(x[2]["ids"]) > MAX_FWD_TOKENS:
+                    capped.append(cur); cur = []
+                cur.append(x)
+            capped.append(cur)
+        for g in capped:
+            items = [it for _, _, it in g]
+            try:
+                probs = await loop.run_in_executor(None, _locked, eng.score_items, items)
+                for (fut, qs, it), p in zip(g, probs):
+                    if not fut.done(): fut.set_result(p)
+            except Exception as e:
+                for fut, _, _ in g:
+                    if not fut.done(): fut.set_exception(e)
+            stats["batches"] += 1; stats["batch_hist"][len(g)] = stats["batch_hist"].get(len(g), 0) + 1
+
+
+@app.on_event("startup")
+async def _start():
+    global eng, queue
+    eng = Engine(MODEL, compile=COMPILE, fp8=FP8, conv_patch=COMPILE); print("[serve] engine", eng.cfg, flush=True)
+    import json
+    global MODEL_NAME, TEMP
+    try: cfg = json.load(open(os.path.join(MODEL, "decider_config.json")))
+    except Exception: cfg = {}
+    eng.neutralize_none = bool(cfg.get("neutralize_none", True)); MODEL_NAME = "decider-" + str(cfg.get("version", "dev"))
+    TEMP = float(os.environ.get("DECIDER_TEMPERATURE", cfg.get("temperature", 1.0)))
+    global RELEASE_DATE; RELEASE_DATE = str(cfg.get("release_date", RELEASE_DATE))
+    global SCHEMA_FIRST, se, squeue, ISOLATED
+    ISOLATED = bool(cfg.get("isolated_levels", False))
+    # the schema cache needs the questions-first layout, which costs accuracy (about 1.5 points on fixed label sets, more on large
+    # label sets and long states): on when the model's config makes it the default, or with DECIDER_SCHEMA_CACHE=1
+    trained = bool(cfg.get("schema_first", False) or cfg.get("schema_first_trained", False))
+    SCHEMA_FIRST = trained and (bool(cfg.get("schema_first", False)) or os.environ.get("DECIDER_SCHEMA_CACHE", "0") == "1")
+    global TEMP_SCHEMA; TEMP_SCHEMA = float(cfg.get("temperature_schema_first", TEMP))
+    if SCHEMA_FIRST:
+        from decider.schema_engine import SchemaEngine
+        se = SchemaEngine(eng); squeue = asyncio.Queue(); asyncio.create_task(schema_batcher()); print("[serve] schema cache on", flush=True)
+        pre = os.environ.get("DECIDER_SCHEMAS")             # JSON file: [{"questions": {...}, "independent": true, "batch_sizes": [1, 8, 32], "state_tokens": [64, 256]}]
+        for spec in (json.load(open(pre)) if pre else []):  # known schemas: prefix computed, graphs compiled and captured before traffic
+            _, h, _ = _schema_handle(spec["questions"], spec.get("independent", True), compile=COMPILE)
+            t = se.warmup(h, spec.get("batch_sizes", (1, 8, 32)), spec.get("state_tokens", (64, 128, 256)))
+            print(f"[serve] preloaded schema with {h.nq} rows, prefix {sum(h.tps)} tokens, graphs ready in {t:.0f}s", flush=True)
+    shapes = [(B, T) for B in (1, 2, 4, 8, 16, 32) for T in T_BUCKETS if T <= eng.max_ctx + 256]
+    if MAX_BATCH > 32: shapes += [(64, T) for T in T_BUCKETS if T <= 512]
+    t = eng.warmup(shapes); print(f"[serve] captured {len(shapes)} graphs in {t:.0f}s", flush=True)
+    queue = asyncio.Queue()
+    asyncio.create_task(batcher())
+
+
+@app.post("/decide")
+async def decide(r: Req):
+    qs, it = await asyncio.get_running_loop().run_in_executor(None, _prepare, r.context, r.schema_)
+    fut = asyncio.get_running_loop().create_future()
+    await queue.put((fut, qs, it))
+    probs = await fut
+    stats["requests"] += 1; stats["decisions"] += len(qs)
+    return _format(r.schema_, qs, probs)
+
+
+SCHEMA_FIRST = False; se = None; squeue = None; schemas = {}            # schema cache (models trained on the questions-first layout, v7+)
+
+
+ISOLATED = False
+
+
+def _schema_handle(questions, independent, compile=False):
+    """Compile (or look up) the question schema: its prefix is run once, requests then only run the state."""
+    import json
+    key = (json.dumps(questions, sort_keys=True, ensure_ascii=False), independent)
+    if key not in schemas:
+        rqs = {k: S1.render_question(v) for k, v in questions.items()}; rows, index = S1.plan_rows(rqs, ISOLATED and independent)
+        with gpu_lock:
+            if len(schemas) >= 128:
+                old = next(iter(schemas)); hid = schemas.pop(old)[1].id
+                for k in [k for k in se.graphs if k[0] == hid]: del se.graphs[k]
+            h = se.prepare(rows, independent=independent, compile=compile)
+        schemas[key] = (rqs, h, index)
+    return schemas[key]
+
+
+seen = {}
+
+
+def _worth_caching(questions, independent):
+    """A schema gets a cached prefix and CUDA graphs from its second request on: one-off schemas go through the generic
+    state-first engine, whose graphs do not depend on the questions, so ad-hoc traffic cannot thrash graph captures."""
+    import json
+    key = (json.dumps(questions, sort_keys=True, ensure_ascii=False), independent)
+    if key in schemas: return True
+    if len(seen) > 50000: seen.clear()
+    seen[key] = seen.get(key, 0) + 1
+    return seen[key] >= int(os.environ.get("DECIDER_SCHEMA_MIN_SEEN", "2"))
+
+
+def _score_schema(h, rows):
+    with gpu_lock:
+        return se.score_rows(h, rows, temperature=TEMP_SCHEMA)
+
+
+async def schema_batcher():
+    """Requests that share a schema and arrive within the window are scored in one forward pass over their states."""
+    loop = asyncio.get_running_loop()
+    while True:
+        batch = await _collect(squeue)
+        groups = {}                                        # one forward per (schema, length bucket): short states are not padded to long ones
+        for fut, h, row in batch: groups.setdefault((h.id, se.bucket(len(row[0]))), (h, []))[1].append((fut, row))
+        for h, items in groups.values():
+            step = max(1, MAX_BATCH // h.P)
+            for i in range(0, len(items), step):
+                chunk = items[i:i + step]
+                try:
+                    probs = await loop.run_in_executor(None, _score_schema, h, [c for _, c in chunk])
+                    for (fut, _), p in zip(chunk, probs):
+                        if not fut.done(): fut.set_result(p)
+                except Exception as e:
+                    for fut, _ in chunk:
+                        if not fut.done(): fut.set_exception(e)
+            stats["schema_batches"] = stats.get("schema_batches", 0) + 1
+
+
+def _locked(fn, items):
+    with gpu_lock:
+        return fn(items, temperature=TEMP)          # fitted temperature from decider_config.json
+
+
+class S1Req(BaseModel):
+    state: object
+    questions: dict
+    model: str | None = None
+    independent: bool = True
+    layout: str | None = None            # "state_first" forces the uncached layout on a schema-first model
+
+
+def _prepare_s1(state, questions, independent):
+    ctx = S1.render_state(state); rqs = {k: S1.render_question(v) for k, v in questions.items()}
+    flat, index = S1.plan_rows(rqs, ISOLATED and independent)
+    rows = [[r] for r in flat] if independent else [flat]
+    items = [build(Example(ctx, [Q(r["question"], list(r["options"]), 0) for r in row]), eng.tok, _NoShuffle(), max_options=MAX_OPTIONS,
+                   max_ctx_tokens=MAX_STATE_TOKENS) for row in rows]
+    return (rqs, index), items
+
+
+@app.post("/v1/systemone")
+async def systemone(r: S1Req):
+    loop = asyncio.get_running_loop()
+    if SCHEMA_FIRST and r.layout != "state_first" and _worth_caching(r.questions, r.independent):
+        try:
+            rqs, h, index = await loop.run_in_executor(None, _schema_handle, r.questions, r.independent)
+        except ValueError as e:
+            raise HTTPException(422, str(e))
+        row = await loop.run_in_executor(None, lambda: se.tokenize(h, S1.render_state(r.state), MAX_STATE_TOKENS))      # CPU work stays off the GPU lock
+        fut = loop.create_future(); await squeue.put((fut, h, row)); p = await fut
+        stats["requests"] += 1; stats["decisions"] += len(rqs); stats["schema_requests"] = stats.get("schema_requests", 0) + 1
+        return {"model": MODEL_NAME, "answers": S1.assemble(rqs, index, [pk.tolist() for pk in p]),
+                "usage": {"input_tokens": len(row[0]) * h.P, "cached_tokens": sum(h.tps), "output_tokens": 0}}
+    try:
+        (rqs, index), items = await loop.run_in_executor(None, _prepare_s1, r.state, r.questions, r.independent)
+    except ValueError as e:
+        raise HTTPException(422, str(e))
+    if len(items) > 1 and min(len(it["ids"]) for it in items) >= SHARED_MIN_TOKENS:
+        res = await loop.run_in_executor(None, _locked, eng.score_shared, items)      # long state: run it once, fork the cache per question
+        stats["shared_prefix_requests"] = stats.get("shared_prefix_requests", 0) + 1
+    else:
+        futs = []
+        for it in items:
+            f = loop.create_future(); futs.append(f); await queue.put((f, None, it))
+        res = await asyncio.gather(*futs)
+    probs = [p for ps in res for p in ps]                                   # one prob row per question, request order
+    stats["requests"] += 1; stats["decisions"] += len(rqs)
+    return {"model": MODEL_NAME, "answers": S1.assemble(rqs, index, [p.tolist() for p in probs]),
+            "usage": {"input_tokens": S1.unique_tokens(items), "output_tokens": 0}}
+
+
+@app.get("/v1/models")
+async def models():
+    return {"models": [{"name": MODEL_NAME, "description": "decider: one-pass typed decisions with calibrated probabilities", "release_date": RELEASE_DATE}]}
+
+
+@app.get("/health")
+async def health():
+    return {"ok": eng is not None, "model": MODEL}
+
+
+@app.get("/stats")
+async def get_stats():
+    return dict(stats, engine=eng.stats if eng else None, graphs=len(eng.graphs) if eng else 0)

decider/systemone.py

@@ -0,0 +1,139 @@
+"""Jev-shaped requests on top of the decider prompt format (same wire format as TypeSafe's POST /v1/systemone).
+
+    state      str | dict | list            JSON state is serialised compactly; questions may name a part by path (`ticket.messages[0].text`)
+    questions  {id: {"type": "choice", "instructions": ..., "criteria": {name: description | {...} | [...] | None}}      up to 255 options
+                     {"type": "score",  "instructions": ..., "criteria": [level 0 description, level 1 description, ...]}  2..10 levels
+                     {"type": "noul",   "instructions": ..., "criteria": {"true": ..., "false": ...} (optional)}}
+    ids are never shown to the model.  `instructions` and every description may be a string or any JSON value.
+"""
+import json, math
+
+MAX_CHOICE, MAX_LEVELS = 255, 10
+
+
+def _txt(v):
+    return v if isinstance(v, str) else json.dumps(v, ensure_ascii=False)
+
+
+ANNOTATE_MIN = 8
+
+
+def annotate_indices(x, min_len=ANNOTATE_MIN):
+    """Write each element's position into long arrays ({"_index": i, ...}).  A path such as `records[47].text` otherwise makes
+    the model count 47 elements; with the index written down it is a lookup (json_k64 probe: 0.49 -> 0.57 accuracy)."""
+    if isinstance(x, list):
+        if len(x) >= min_len:
+            return [({"_index": i, **annotate_indices(v, min_len)} if isinstance(v, dict) else {"_index": i, "value": annotate_indices(v, min_len)}) for i, v in enumerate(x)]
+        return [annotate_indices(v, min_len) for v in x]
+    if isinstance(x, dict):
+        return {k: annotate_indices(v, min_len) for k, v in x.items()}
+    return x
+
+
+def render_state(state, index_arrays=True):
+    if isinstance(state, str):
+        return state
+    return json.dumps(annotate_indices(state) if index_arrays else state, ensure_ascii=False)
+
+
+def render_question(spec):
+    """-> dict(question=str, options=[str], type=..., names=[...])  (names: what the answer reports for each option)"""
+    t = spec.get("type", "choice"); ins = _txt(spec.get("instructions", spec.get("question", ""))); crit = spec.get("criteria", spec.get("options"))
+    if not ins:
+        raise ValueError("question without instructions")
+    if t == "choice":
+        if isinstance(crit, (list, tuple)):
+            crit = {str(c): None for c in crit}
+        if not isinstance(crit, dict) or not 2 <= len(crit) <= MAX_CHOICE:
+            raise ValueError(f"choice criteria: a map of 2..{MAX_CHOICE} options")
+        names = list(crit); opts = [n if crit[n] in (None, "") else f"{n}: {_txt(crit[n])}" for n in names]
+    elif t == "score":
+        if isinstance(crit, dict):                                   # legend form {"0": "...", "1": "..."}
+            crit = [crit[k] for k in sorted(crit, key=float)]
+        if not isinstance(crit, (list, tuple)) or not 2 <= len(crit) <= MAX_LEVELS:
+            raise ValueError(f"score criteria: an ordered list of 2..{MAX_LEVELS} level descriptions")
+        names = list(range(len(crit))); opts = [f"{i}: {_txt(c)}" for i, c in enumerate(crit)]
+    elif t in ("noul", "bool"):
+        names = [False, True]; c = crit or {}
+        f, tr = c.get("false", c.get(False)), c.get("true", c.get(True))
+        opts = ["no" if f in (None, "") else f"no: {_txt(f)}", "yes" if tr in (None, "") else f"yes: {_txt(tr)}"]
+    else:
+        raise ValueError(f"unknown question type {t!r}")
+    return dict(question=ins, options=opts, type="noul" if t == "bool" else t, names=names, legend=[_txt(c) for c in crit] if t == "score" else None,
+                isolated=bool(spec.get("isolated", True)))
+
+
+# ---- isolated levels: every Score level is judged in its own row, without its number or its neighbours
+ISOLATED = "{q}\nProposed answer: {level}\nDoes the proposed answer fit?"
+_NUM = None
+
+
+def strip_level_number(text):
+    """"2: somewhat" -> "somewhat" (dataset legends carry the number; an isolated level must not)."""
+    import re
+    return re.sub(r"^\s*-?\d+\s*:\s*", "", text)
+
+
+def isolated_rows(question, levels):
+    """-> one yes/no question per level: [(question text, ["no", "yes"])]."""
+    return [(ISOLATED.format(q=question, level=strip_level_number(l)), ["no", "yes"]) for l in levels]
+
+
+def combine_isolated(p_yes):
+    """Per-level P(fits), each computed without reference to any other level -> a distribution over levels.
+    Also returns the unnormalised mass: near 1 when exactly one level fits, low when none does, high when several do."""
+    tot = sum(p_yes) or 1e-9
+    return [x / tot for x in p_yes], tot
+
+
+def plan_rows(rqs, isolated=True):
+    """One scoring row per question; a Score question with isolated levels becomes one yes/no row per level.
+    -> (rows [{"question", "options"}], index [(id, "iso" | "list", first row, n rows)])"""
+    rows, index = [], []
+    for k, r in rqs.items():
+        if isolated and r["type"] == "score" and r.get("isolated", True):
+            rws = isolated_rows(r["question"], r["legend"]); index.append((k, "iso", len(rows), len(rws))); rows += [dict(question=t, options=o) for t, o in rws]
+        else:
+            index.append((k, "list", len(rows), 1)); rows.append(dict(question=r["question"], options=r["options"]))
+    return rows, index
+
+
+def assemble(rqs, index, probs):
+    """probs: one probability list per row (plan_rows order) -> {id: answer}."""
+    out = {}
+    for k, kind, s, n in index:
+        if kind == "iso":
+            fit = [float(probs[s + j][1]) for j in range(n)]; p, mass = combine_isolated(fit); a = format_answer(rqs[k], p)
+            a["level_fit"] = {str(j): round(x, 4) for j, x in enumerate(fit)}; a["fit_mass"] = round(mass, 4); out[k] = a
+        else:
+            out[k] = format_answer(rqs[k], probs[s])
+    return out
+
+
+def certainty(p):
+    """1 - normalised entropy: 1 when all mass is on one option, 0 when the distribution is flat."""
+    h = -sum(x * math.log(x) for x in p if x > 0)
+    return max(0.0, 1.0 - h / math.log(len(p))) if len(p) > 1 else 1.0
+
+
+def format_answer(rq, p, nd=4):
+    """rq: render_question output; p: probabilities in option order."""
+    p = [float(x) for x in p[:len(rq["options"])]]; s = sum(p) or 1.0; p = [x / s for x in p]
+    j = max(range(len(p)), key=p.__getitem__)
+    if rq["type"] == "noul":
+        return {"type": "noul", "noul": round(p[1], nd)}
+    if rq["type"] == "choice":
+        return {"type": "choice", "choice": rq["names"][j], "confidence": round(p[j], nd), "certainty": round(certainty(p), nd),
+                "probabilities": {n: round(x, nd) for n, x in zip(rq["names"], p)}}
+    return {"type": "score", "score": round(sum(i * x for i, x in enumerate(p)), 2), "confidence": round(p[j], nd), "certainty": round(certainty(p), nd),
+            "legend": {str(i): d for i, d in enumerate(rq["legend"])}, "probabilities": {str(i): round(x, nd) for i, x in enumerate(p)}}
+
+
+def unique_tokens(items):
+    """Input tokens of a request whose rows share a prefix (the state): the prefix counts once."""
+    ids = [it["ids"] for it in items]
+    if len(ids) < 2:
+        return sum(len(x) for x in ids)
+    lcp = 0; short = min(len(x) for x in ids)
+    while lcp < short and all(x[lcp] == ids[0][lcp] for x in ids): lcp += 1
+    return lcp + sum(len(x) - lcp for x in ids)

decider_config.json

@@ -0,0 +1 @@
+{"temperature": 1.03, "temperature_schema_first": 1.03, "neutralize_none": false, "version": "0.8b-v1", "base": "Qwen/Qwen3.5-0.8B-Base", "max_options": 255, "max_state_tokens": 32768, "schema_first": false, "schema_first_trained": true, "isolated_levels": true, "recipe": "scripts/train.sh full (single run)", "release_date": "2026-09-19"}

generation_config.json

@@ -0,0 +1,6 @@
+{
+  "_from_model_config": true,
+  "eos_token_id": 248044,
+  "transformers_version": "5.17.0",
+  "use_cache": true
+}

model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6926f82ef7e9ea408ad881555204daa6bb694ca82509c4d1514b7be2e713563d
+size 1504827608

tokenizer.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:06b9509352d2af50381ab2247e083b80d32d5c0aba91c272ca9ff729b6a0e523
+size 19989325

tokenizer_config.json

@@ -0,0 +1,32 @@
+{
+  "add_prefix_space": false,
+  "audio_bos_token": "<|audio_start|>",
+  "audio_eos_token": "<|audio_end|>",
+  "audio_token": "<|audio_pad|>",
+  "backend": "tokenizers",
+  "bos_token": null,
+  "clean_up_tokenization_spaces": false,
+  "eos_token": "<|endoftext|>",
+  "errors": "replace",
+  "image_token": "<|image_pad|>",
+  "is_local": false,
+  "local_files_only": false,
+  "model_max_length": 262144,
+  "model_specific_special_tokens": {
+    "audio_bos_token": "<|audio_start|>",
+    "audio_eos_token": "<|audio_end|>",
+    "audio_token": "<|audio_pad|>",
+    "image_token": "<|image_pad|>",
+    "video_token": "<|video_pad|>",
+    "vision_bos_token": "<|vision_start|>",
+    "vision_eos_token": "<|vision_end|>"
+  },
+  "pad_token": "<|endoftext|>",
+  "pretokenize_regex": "(?i:'s|'t|'re|'ve|'m|'ll|'d)|[^\\r\\n\\p{L}\\p{N}]?[\\p{L}\\p{M}]+|\\p{N}| ?[^\\s\\p{L}\\p{M}\\p{N}]+[\\r\\n]*|\\s*[\\r\\n]+|\\s+(?!\\S)|\\s+",
+  "split_special_tokens": false,
+  "tokenizer_class": "Qwen2Tokenizer",
+  "unk_token": null,
+  "video_token": "<|video_pad|>",
+  "vision_bos_token": "<|vision_start|>",
+  "vision_eos_token": "<|vision_end|>"
+}