"""Rendering, candidate aggregation and service logic, without model imports."""
from __future__ import annotations

import json
import math
import re
import string
from pathlib import Path

import threading
if __package__:
    from .backends import new_meter, check_deadline
    from .quota import QuotaStateError
else:
    from backends import new_meter, check_deadline
    from quota import QuotaStateError

ROOT = Path(__file__).resolve().parent
YES_FORMS = ["yes", "Yes", "YES", " yes", " Yes", " YES", "true", "True", " true", " True", "Y", " Y"]
NO_FORMS = ["no", "No", "NO", " no", " No", " NO", "false", "False", " false", " False", "N", " N"]
ALIASES = {"boolean": "noul", "enum": "choice", "noul": "noul", "choice": "choice", "score": "score"}


class ItemFailure(RuntimeError):
    """Inference for one item failed; the service answers HTTP 422 and keeps running."""


class ServiceFailure(RuntimeError):
    """Service-level failure; the service answers HTTP 503."""


# The third consecutive item failure is treated as service-level (HTTP 503, service
# marked failed), so the harness stop rule for consecutive infrastructure errors applies.
ITEM_FAILURE_LIMIT = 3


def unrecoverable_engine_error(exc):
    """True if the __cause__/__context__ graph of exc holds an unrecoverable vLLM engine error.

    Classes are matched by name, so the check needs no vLLM import: EngineDeadError, or a
    vLLM VLLMServerError subclass other than the per-request EngineGenerateError.
    """
    pending, seen = [exc], set()
    while pending:
        current = pending.pop()
        if current is None or id(current) in seen:
            continue
        seen.add(id(current))
        classes = type(current).__mro__
        if isinstance(current, ServiceFailure) or any(cls.__name__ == 'EngineDeadError' for cls in classes):
            return True
        vllm_classes = {cls.__name__ for cls in classes
                        if cls.__module__ == 'vllm' or cls.__module__.startswith('vllm.')}
        if 'VLLMServerError' in vllm_classes and 'EngineGenerateError' not in vllm_classes:
            return True
        pending.extend((current.__cause__, current.__context__))
    return False


def external_path(path):
    """Resolve an absolute path that is neither inside the package nor an ancestor of it."""
    candidate = Path(path)
    if not candidate.is_absolute():
        raise ValueError("Path must be absolute: " + str(path))
    result = candidate.resolve()
    if result == ROOT or result.is_relative_to(ROOT) or ROOT.is_relative_to(result):
        raise ValueError("Path must be outside the package and must not contain it: " + str(result))
    return result


def external_output(path):
    """Validated output location outside the package; callers create files with mode 'x'."""
    result = external_path(path)
    result.parent.mkdir(parents=True, exist_ok=True)
    return result


def load_json(path):
    def reject(value):
        raise ValueError("Non-finite JSON value")
    return json.loads(Path(path).read_text(encoding="utf-8"), parse_constant=reject)


def load_config(path=None):
    config = load_json(path or ROOT / "config.json")
    apply_profile(config)
    validate_config(config)
    return config


def apply_profile(config):
    import copy
    profile=config.get('profiles',{}).get(config['model']['model_key'])
    if profile is None: raise ValueError('No profile for model_key')
    for section,values in profile.items():
        config.setdefault(section,{}).update(copy.deepcopy(values))
    return apply_budget_tier(config)


def apply_budget_tier(config, tier=None):
    r=config['routing'];tier=tier or r.get('tier','normal')
    if tier not in ('normal','hard','long'):raise ValueError('Unknown quota tier')
    r['tier']=tier
    tiers=r.get('tiers',{})
    values=tiers.get(tier,{})
    for key in ('reasoning_words','max_new_tokens'):
        if key in tiers.get('hard',{}):config['generation'][key]=tiers['hard'][key]
    if 'max_prompt_tokens' in values:config['limits']['max_prompt_tokens']=values['max_prompt_tokens']
    for key in ('reasoning_words','max_new_tokens'):
        if key in values:config['generation'][key]=values[key]
    # Forced long is a runtime admission switch, preserving fast identity.
    return config


def finite_number(value, name, *, low=None, high=None, positive=False):
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
        raise ValueError(f"{name} must be a finite number")
    if (low is not None and value < low) or (high is not None and value > high) or (positive and value <= 0):
        raise ValueError(f"{name} is outside its allowed range")
    return value


def validate_config(c):
    if __package__: from .quota import cost_settings
    else: from quota import cost_settings
    cost_settings(c)
    if c['model']['model_key'] not in c.get('profiles',{}):raise ValueError('No profile for model_key')
    if c.get('usage',{}).get('basis','engine') not in ('engine','each_once','submitted'):
        raise ValueError('Invalid usage basis')
    m, r, g, p, s = (c[k] for k in ("model", "routing", "generation", "prompt", "server"))
    if m["backend"] not in ("transformers", "vllm") or m["quant"] not in ("int8", "bf16", "awq"):
        raise ValueError("Unsupported backend or precision")
    if m['backend']=='vllm':
        neutral={'suppress_tokens':None,'begin_suppress_tokens':None,'bad_words_ids':None,
            'forced_bos_token_id':None,'forced_eos_token_id':None,'forced_decoder_ids':None,
            'no_repeat_ngram_size':0,'min_new_tokens':0,'min_length':0,'length_penalty':1.0,'renormalize_logits':False}
        if any(g.get(k,v)!=v for k,v in neutral.items()):raise ValueError('Unsupported vLLM decode override')
    if r["mode"] not in ("routed", "fast_only", "uniform"):
        raise ValueError("Unsupported routing mode")
    tiers=r.get('tiers',{})
    if tiers:
        if set(tiers)!={'normal','hard','long'} or r.get('tier','normal') not in tiers:raise ValueError('Require normal/hard/long tiers')
        if tiers['long'].get('fast_only') is not True:raise ValueError('Long tier must be fast only')
        for tier in ('normal','long'):
            value=tiers[tier].get('max_prompt_tokens')
            if type(value) is not int or not 0<value<=c['limits']['physical_max_prompt_tokens']:raise ValueError('Invalid tier input budget')
        for key in ('reasoning_words','max_new_tokens'):
            if type(tiers['hard'].get(key)) is not int or tiers['hard'][key]<=0:raise ValueError('Invalid hard draft budget')
    finite_number(r["margin_threshold"], "margin_threshold", low=0)
    if r["margin_kind"]=="probdiff" and r["margin_threshold"]>1:
        raise ValueError("probdiff threshold must be <=1")
    if r["margin_kind"] not in ("logratio", "probdiff"):
        raise ValueError("Invalid margin_kind")
    if r.get('long_item_slow_policy','fast_only') not in ('fast_only','allow'):
        raise ValueError('Invalid long item policy')
    if r["unparsed_policy"] not in ("fast", "force_answer"):
        raise ValueError("Invalid unparsed_policy")
    finite_number(r["quota"], "quota", low=0, high=1)
    finite_number(r["fuse"], "fuse", low=0, high=1)
    if r["quota"] > r["fuse"]:
        raise ValueError("quota must not exceed fuse")
    if r["mode"] == "uniform" and (r["quota"] != 1 or r["fuse"] != 1):
        raise ValueError("uniform requires explicit quota=1 and fuse=1; it is an unbounded draft experiment")
    for key in ("reasoning_words", "max_new_tokens"):
        if type(g[key]) is not int or g[key] <= 0:
            raise ValueError(f"{key} must be a positive integer")
    if type(g["seed"]) is not int or type(g["do_sample"]) is not bool:
        raise ValueError("Invalid generation seed or do_sample")
    finite_number(g.get("wall_timeout_seconds",60),"slow wall timeout",positive=True)
    finite_number(g["temperature"], "generation temperature", positive=True)
    finite_number(g["top_p"], "top_p", positive=True, high=1)
    finite_number(g["repetition_penalty"], "repetition_penalty", positive=True)
    if type(g["top_k"]) is not int or g["top_k"] < 0:
        raise ValueError("top_k must be a nonnegative integer")
    for key in ("max_request_bytes", "max_questions", "port"):
        if type(s[key]) is not int or s[key] <= 0:
            raise ValueError(f"{key} must be a positive integer")
    if s["port"] > 65535:
        raise ValueError("Invalid port")
    finite_number(s["socket_timeout_seconds"], "socket timeout", positive=True)
    if not isinstance(p["answer_marker"], str) or not p["answer_marker"].strip() or "\n" in p["answer_marker"]:
        raise ValueError("answer_marker must be a nonempty single line")
    if type(p["enable_thinking"]) is not bool or not isinstance(p["system"], str):
        raise ValueError("Invalid prompt settings")
    for template in p["cot_instructions"].values():
        template.format(reasoning_words=g["reasoning_words"])
    for value in (c["limits"]["max_prompt_tokens"], r["slow_max_prompt_tokens"]):
        if type(value) is not int or value <= 0:
            raise ValueError("Token limits must be positive integers")
    limits=c['limits']
    for name in ('physical_max_prompt_tokens','oom_retry_prompt_tokens','state_chunk_tokens','precompression_trigger_chars_per_token','estimated_chars_per_token'):
        if type(limits[name]) is not int or limits[name]<=0: raise ValueError('Invalid compression limit: '+name)
    for name in ('state_head_tokens','state_tail_tokens','compression_reserve_tokens'):
        if type(limits[name]) is not int or limits[name]<0: raise ValueError('Invalid compression limit: '+name)
    if limits['physical_max_prompt_tokens']<limits['max_prompt_tokens']:
        raise ValueError('Physical maximum must cover soft prompt budget')
    finite_number(c["limits"]["truncation_head_frac"], "head fraction", low=0, high=1)
    if c.get("interface", {}).get("boolean_keys", "request") not in ("request", "true_false", "yes_no"):
        raise ValueError("Unknown boolean key style")


class UnsupportedRequest(ValueError):
    """Structurally valid request outside the supported semantic domain (422)."""


def choice_codes(count):
    if count<=26:return list(string.ascii_uppercase[:count])
    if count>676:raise UnsupportedRequest('At most 676 choices supported')
    return [a+b for a in string.ascii_uppercase for b in string.ascii_uppercase][:count]


def normalize_question(question):
    if not isinstance(question, dict):
        raise ValueError("Each question must be an object")
    qtype = ALIASES.get(question.get("type"))
    if qtype is None:
        raise UnsupportedRequest("Unsupported question type")
    instructions = question.get("instructions", question.get("description"))
    if not isinstance(instructions, str) or not instructions.strip():
        raise ValueError("Question needs instructions or description")
    criteria = question.get("criteria")
    q = {"request_type": question["type"], "type": qtype, "instructions": instructions, "criteria": criteria}
    labels = None
    if qtype == "noul":
        if criteria is not None and not isinstance(criteria, dict):
            raise ValueError("noul criteria must be an object")
    elif qtype == "score":
        if not isinstance(criteria,list): raise ValueError("score criteria must be an ordered list")
        if not criteria: raise UnsupportedRequest("score criteria is empty")
    else:
        labels = question.get("labels")
        if labels is None:
            labels = list(criteria) if isinstance(criteria, (dict, list)) else None
        if not isinstance(labels,list): raise ValueError("choice needs labels or ordered criteria")
        if not 1<=len(labels)<=676: raise UnsupportedRequest("choice needs 1..676 labels")
        if any(not isinstance(x, str) or not x for x in labels) or len(set(labels)) != len(labels):
            raise ValueError("choice labels must be distinct nonempty strings")
        if criteria is not None and not isinstance(criteria, (dict, list)):
            raise ValueError("choice criteria must be an object or list")
    return q, labels


def normalize_request(payload, max_questions):
    if not isinstance(payload, dict) or "state" not in payload:
        raise ValueError("Request needs state and questions")
    questions = payload.get("questions")
    if not isinstance(questions, dict) or not 1 <= len(questions) <= max_questions:
        raise ValueError("questions must be a nonempty object within max_questions")
    tasks = []
    for name, question in questions.items():
        if not isinstance(name, str) or not name:
            raise ValueError("Question names must be nonempty strings")
        q, labels = normalize_question(question)
        tasks.append((name, {"state": payload["state"], "question": q, "labels": labels}))
    return tasks


def letter_forms(c):
    return [c, " " + c, c.lower(), " " + c.lower(), c + ".", "(" + c]


def digit_forms(d):
    return [d, " " + d, d + "."]


def render(task):
    # Source-faithful copy of cache_hidden_v2.render; do not deduplicate forms.
    q = task["question"]; qtype = q["type"]; crit = q.get("criteria")
    state = task["state"] if isinstance(task["state"], str) else json.dumps(task["state"], ensure_ascii=False)
    cq = "Context:\n" + state + "\n\nQuestion: " + q["instructions"]
    if qtype == "noul":
        labels = ["no", "yes"]; c = crit or {}; groups = {"yes": YES_FORMS, "no": NO_FORMS}
        parts = [cq] + (["Answer yes if: " + str(c.get("true", "Yes")) + "\nAnswer no if: " + str(c.get("false", "No"))] if (c.get("true") or c.get("false")) else []) + ["Answer with one word: yes or no."]
    elif qtype == "score":
        labels = [str(i) for i in range(len(crit))]; groups = {l: digit_forms(l) for l in labels}
        parts = [cq, "Levels:\n" + "\n".join(str(i) + ": " + str(t) for i, t in enumerate(crit)), "Answer with one number from 0 to " + str(len(crit) - 1) + "."]
    else:
        labels = list(task["labels"]); rubric = crit if isinstance(crit, dict) else {}; codes = choice_codes(len(labels))
        groups = {l: letter_forms(c) for c, l in zip(codes, labels)}
        parts = [cq, "Options:\n" + "\n".join(c + ". " + str(l) + ((": " + str(rubric[l])) if rubric.get(l) else "") for c, l in zip(codes, labels)), "Answer with the letter of the best option."]
    return labels, "\n\n".join(parts), groups


def parse_answer(generation, qtype="noul", labels=None):
    """Audit only: keep the explicit-last and standalone fallbacks of the measured parser."""
    if qtype == "noul":
        matches = re.findall(r"ANSWER:\s*(yes|no)", generation, re.I) or re.findall(r"\b(yes|no)\b", generation, re.I)
        return matches[-1].lower() if matches else None
    if qtype == "choice":
        mapping = dict(zip(choice_codes(len(labels or [])), labels or []))
        explicit, fallback = r"ANSWER:\s*([A-Z]{1,2})\b", r"\b([A-Z]{1,2})\b"
    elif qtype == "score":
        mapping = {str(label): str(label) for label in (labels or [])}
        explicit, fallback = r"ANSWER:\s*(\d+)", r"(?<![\w.+-])(\d+)(?!\w|\.\d)"
    else:
        raise ValueError("Unsupported question type")
    matches = re.findall(explicit, generation)
    if matches and matches[-1] in mapping:
        return mapping[matches[-1]]
    matches = [symbol for symbol in re.findall(fallback, generation) if symbol in mapping]
    return mapping[matches[-1]] if matches else None


def readout_prefix(generation, marker):
    matches = list(re.finditer(r"(?m)^[ \t]*" + re.escape(marker), generation))
    if matches:
        return generation[:matches[-1].end()], False
    return generation.rstrip() + "\n" + marker, True


def logsumexp(values):
    top = max(values)
    if top == -math.inf: return top
    return top + math.log(sum(math.exp(x - top) for x in values))


def candidate_ids(backend, groups):
    # Keep multiplicity even when two textual forms map to the same token id.
    result = {}
    for label, forms in groups.items():
        tokens = [backend.encode(form) for form in forms]
        result[label] = [ids[0] for ids in tokens if len(ids) == 1]
    if not any(result.values()):
        raise ValueError("Tokenizer has no single-token candidate forms")
    return result


def aggregate(logits, groups):
    for token in {t for ids in groups.values() for t in ids}:
        if token not in logits or not math.isfinite(logits[token]):
            raise RuntimeError("Backend omitted a candidate or returned a non-finite logit")
    return {label: logsumexp([logits[t] for t in ids]) if ids else -1e9 for label, ids in groups.items()}


def validate_calibration(calibration, model_key, path="fast"):
    model = calibration.get("models", {}).get(model_key)
    if model is None:
        raise ValueError("Calibration lacks selected model_key")
    block = model.get(path)
    if block is None:
        raise ValueError("Calibration lacks selected route")
    for kind in ("choice", "score"):
        finite_number(block["temperature_by_kind"][kind], kind + " temperature", positive=True)
    if block["temperature_by_kind"]["noul"] != "one_bin":
        raise ValueError("noul calibration must be one_bin")
    finite_number(block["noul_calibration"]["p_cal"], "p_cal", low=0, high=1)
    return block


def probabilities(labels, logits, qtype, block):
    if qtype == "noul":
        # Deterministic tie: earliest in render's label order, namely no.
        winner = max(labels, key=logits.__getitem__)
        p_cal = min(1 - 1e-6, max(0.5, block["noul_calibration"]["p_cal"]))
        return {label: p_cal if label == winner else 1 - p_cal for label in labels}
    temperature = block["temperature_by_kind"][qtype]
    # Subtract before dividing to avoid overflow from small positive T.
    top = max(logits.values())
    weights = {label: math.exp((logits[label] - top) / temperature) for label in labels}
    total = sum(weights.values())
    floored = {label: max(1e-6, weights[label] / total) for label in labels}
    return {label: value / sum(floored.values()) for label, value in floored.items()}


# Rule sections are protected conservatively. Unknown text outside a recognized
# section is narrative; explicit rule/policy/definition/criteria subtrees are sacred.
_RULE = re.compile(r'(?im)^\s*(?:R\d+\s*[:：]|(?:rules?|polic(?:y|ies)|definitions?|criteria|规则|規則|定义|定義|政策|判定标准)\s*[:：])')

_RULE_KEY = re.compile(r'^\s*(?:rules?|polic(?:y|ies)|definitions?|criteria|application_rule_set|规则|規則|定义|定義|政策|标准|判定标准)\s*$', re.I)

def contains_rule_subtree(value):
    if isinstance(value,str):return bool(_RULE.search(value))
    if isinstance(value,dict):return any(_RULE_KEY.fullmatch(str(k)) or contains_rule_subtree(v) for k,v in value.items())
    if isinstance(value,list):return any(contains_rule_subtree(v) for v in value)
    return False


OMISSION = '\n[... state middle omitted ...]\n'


def lexical_terms(text):
    words = re.findall(r'[a-zA-Z_][\w-]*|\d+(?:\.\d+)?|[\u4e00-\u9fff]', text.lower())
    # Character bigrams preserve Chinese query signal without a segmenter.
    words += re.findall(r'(?=([\u4e00-\u9fff]{2}))', text)
    return set(words)


def shrink_state(state, keep_chars, head_fraction=0.6, *, backend=None, query='',
                 head_tokens=512, tail_tokens=512, chunk_tokens=200):
    """Extract query-relevant chunks in original order, protecting explicit rules.

    The budget is tokens when a backend is supplied, otherwise characters.
    A contiguous omitted range produces exactly one marker.
    """
    encode = backend.encode if backend else lambda x: list(x)
    decode = backend.decode if backend else lambda x: ''.join(x)
    query_terms = lexical_terms(query)
    important = {w for w in query_terms if any(c.isdigit() for c in w)}
    important |= {w.lower() for w in re.findall(r'\b(?:[A-Z][a-z]+|[A-Z][A-Z0-9_]+)\b',query)}
    def text_shrink(text, budget):
        if len(encode(text)) <= budget:
            return text
        # Slice original paragraphs including their delimiters; never make a
        # delimiter into an independent compressible paragraph.
        spans=[]; rule_section=False
        for match in re.finditer(r'.+?(?:\n[ \t]*\n|\Z)',text,re.S):
            part=match.group()
            heading=re.match(r'^\s*(?:#{1,6}\s*)?([^\n:：]{1,60})[:：]\s*(?:\n|$)',part)
            if heading:
                title=heading.group(1)
                if _RULE_KEY.search(title): rule_section=True
                elif re.fullmatch(r'(?i)narrative|events?|history|state|context|data|observations?|背景|叙述|事件|状态|记录',title.strip()): rule_section=False
            protected=rule_section or bool(_RULE.search(part))
            if spans and spans[-1][1]==protected: spans[-1]=(spans[-1][0]+part,protected)
            else: spans.append((part,protected))
        merged=spans
        encoded=[(part,keep,encode(part)) for part,keep in merged]
        total=sum(len(ids) for _,_,ids in encoded)
        head_end=min(head_tokens,total);tail_start=max(head_end,total-tail_tokens)
        blocks=[];cursor=0
        for part,protected,ids in encoded:
            if protected:
                blocks.append((part,len(ids),True));cursor+=len(ids)
                continue
            # Sentence/newline boundaries first, word boundaries for oversized
            # sentences. Never decode an arbitrary token slice into a block.
            sentences=re.findall(r'.*?(?:[.!?。！？](?=\s|$)|\n|$)',part,re.S)
            pieces=[]
            for sentence in sentences:
                if not sentence:continue
                if len(encode(sentence))<=chunk_tokens:pieces.append(sentence)
                else:
                    words=re.findall(r'\s+|\S+\s*',sentence)
                    chunk=''
                    for word in words:
                        if chunk and len(encode(chunk+word))>chunk_tokens:
                            pieces.append(chunk);chunk=''
                        chunk+=word
                    if chunk:pieces.append(chunk)
            chunk=''
            def add(value):
                nonlocal cursor
                size=len(encode(value))
                edge=cursor<head_end or cursor+size>tail_start
                blocks.append((value,size,edge));cursor+=size
            for piece in pieces:
                if chunk and len(encode(chunk+piece))>chunk_tokens:
                    add(chunk);chunk=''
                chunk+=piece
            if chunk:add(chunk)
        selected={i for i,(_,_,keep) in enumerate(blocks) if keep}
        marker_cost=len(encode(OMISSION))
        used=sum(blocks[i][1] for i in selected)+marker_cost*(len(selected)+1)
        ranked=[]
        for i,(part,size,keep) in enumerate(blocks):
            terms=lexical_terms(part); overlap=query_terms & terms
            score=(len(overlap)+2*len(important & terms))/max(1,math.sqrt(len(terms)))
            if not keep: ranked.append((-score,i))
        for _,i in sorted(ranked):
            if used+blocks[i][1]+marker_cost <= budget:
                selected.add(i); used+=blocks[i][1]+marker_cost
        def join(indices):
            out=[]; omitted=False
            for i,(part,_,_) in enumerate(blocks):
                if i in indices:
                    if omitted: out.append(OMISSION)
                    out.append(part); omitted=False
                else: omitted=True
            if omitted: out.append(OMISSION)
            return ''.join(out)
        result=join(selected)
        # Recover conservative omission-marker reserves using adjacent blocks.
        while len(encode(result))<.8*budget:
            neighbors=sorted({j for i in selected for j in (i-1,i+1) if 0<=j<len(blocks) and j not in selected})
            added=False
            for j in neighbors:
                candidate=join(selected|{j})
                if len(encode(candidate))<=budget:
                    selected.add(j);result=candidate;added=True
            if not added:break
        if len(encode(result)) <= budget: return result
        # Protected material survives the soft budget. For an unusually small
        # budget shrink only narrative edges globally, retaining the rules.
        if any(keep for _,keep in merged):
            protected_text=''.join(part for part,keep in merged if keep)
            allowance=max(0,budget-len(encode(protected_text))-marker_cost*(len(merged)+1))
            if not allowance: return protected_text
            n=sum(len(encode(part)) for part,keep in merged if not keep)
            h=int(allowance*head_fraction);t=allowance-h
            out=[];offset=0;omitted=False
            for part,keep in merged:
                if keep:
                    if omitted:out.append(OMISSION);omitted=False
                    out.append(part);continue
                ids=encode(part)
                head=ids[:max(0,min(len(ids),h-offset))]
                tail=ids[max(0,n-t-offset):] if offset+len(ids)>n-t else []
                if head:
                    if omitted:out.append(OMISSION);omitted=False
                    out.append(decode(head))
                if len(head)+len(tail)<len(ids):omitted=True
                if tail:
                    if omitted:out.append(OMISSION);omitted=False
                    out.append(decode(tail))
                offset+=len(ids)
            if omitted:out.append(OMISSION)
            return ''.join(out)
        ids=encode(text); available=max(0,budget-marker_cost)
        h=int(available*head_fraction); t=available-h
        return decode(ids[:h])+OMISSION+(decode(ids[-t:]) if t else '')

    if isinstance(state,str): return text_shrink(state,max(0,keep_chars))
    if isinstance(state,dict):
        protected={k:v for k,v in state.items() if _RULE_KEY.search(str(k)) or
                   (not isinstance(v,(dict,list)) and len(str(v))<=200)}
        rest=[k for k in state if k not in protected]
        remaining=max(0,keep_chars-len(encode(json.dumps(protected,ensure_ascii=False))))
        sizes={k:len(encode(json.dumps(state[k],ensure_ascii=False))) for k in rest}
        total=max(1,sum(sizes.values()))
        return {k:v if k in protected else shrink_state(v,int(remaining*sizes[k]/total),head_fraction,
                backend=backend,query=query,head_tokens=head_tokens,tail_tokens=tail_tokens,chunk_tokens=chunk_tokens)
                for k,v in state.items()}
    if isinstance(state,list):
        if len(encode(json.dumps(state,ensure_ascii=False)))<=keep_chars: return state
        if not state: return state
        # Budget whole entries, protect explicit rule entries; preserve both ends.
        selected={0,len(state)-1}|{i for i,v in enumerate(state) if contains_rule_subtree(v)}
        ranked=sorted(range(1,len(state)-1),key=lambda i:(-len(lexical_terms(str(state[i])) & query_terms),i))
        used=sum(len(encode(json.dumps(state[i],ensure_ascii=False))) for i in selected)
        for i in ranked:
            size=len(encode(json.dumps(state[i],ensure_ascii=False)))
            if i not in selected and used+size+len(OMISSION)<keep_chars: selected.add(i); used+=size
        protected_entries={i for i in selected if isinstance(state[i],str) and _RULE.search(state[i])}
        protected_size=sum(len(encode(json.dumps(state[i],ensure_ascii=False))) for i in protected_entries)
        narrative_size=max(1,sum(len(encode(json.dumps(state[i],ensure_ascii=False))) for i in selected-protected_entries))
        free=max(0,keep_chars-protected_size-len(selected)*len(OMISSION))
        result=[]; omitted=False
        for i,v in enumerate(state):
            if i in selected:
                if omitted: result.append(OMISSION.strip())
                if i not in protected_entries and len(str(v))>200:
                    size=len(encode(json.dumps(v,ensure_ascii=False)))
                    v=shrink_state(v,int(free*size/narrative_size),head_fraction,backend=backend,
                        query=query,head_tokens=head_tokens,tail_tokens=tail_tokens,chunk_tokens=chunk_tokens)
                result.append(v);omitted=False
            else: omitted=True
        return result
    return state


def prepare_task(task, backend, config):
    def prompts(value):
        labels,text,groups=render(value)
        instruction=config['prompt']['cot_instructions'][value['question']['type']].format(
            reasoning_words=config['generation']['reasoning_words']).replace('ANSWER:',config['prompt']['answer_marker'])
        system,think=config['prompt']['system'],config['prompt']['enable_thinking']
        return labels,groups,backend.prompt_ids(system,text,think),backend.prompt_ids(system,text+'\n'+instruction,think)
    limits=config['limits']; budget=limits['max_prompt_tokens'];normal_budget=budget
    long_budget=min(budget,config['routing'].get('tiers',{}).get('long',{}).get('max_prompt_tokens',budget))
    query=task['question']['instructions']+' '+json.dumps(task['question'].get('criteria'),ensure_ascii=False)+' '+str(task.get('labels') or '')
    raw=task['state'] if isinstance(task['state'],str) else json.dumps(task['state'],ensure_ascii=False)
    precompressed=len(raw)>limits['precompression_trigger_chars_per_token']*budget
    source=task
    if precompressed:budget=long_budget
    if precompressed:
        ratio=limits['estimated_chars_per_token']
        source=dict(task,state=shrink_state(task['state'],budget*ratio,limits['truncation_head_frac'],query=query,
            head_tokens=limits['state_head_tokens']*ratio,tail_tokens=limits['state_tail_tokens']*ratio,
            chunk_tokens=limits['state_chunk_tokens']*ratio))
    original=prompts(source); original_count=len(original[2])
    long_input=precompressed or original_count>normal_budget
    if long_input:budget=long_budget
    current,prepared=source,original
    refilled=False
    if precompressed and original_count<.8*budget:
        fixed=len(prompts(dict(task,state=''))[2])
        refill=shrink_state(task['state'],max(0,budget-fixed-limits['compression_reserve_tokens']),
            limits['truncation_head_frac'],backend=backend,query=query,
            head_tokens=limits['state_head_tokens'],tail_tokens=limits['state_tail_tokens'],chunk_tokens=limits['state_chunk_tokens'])
        candidate=dict(task,state=refill);candidate_prompts=prompts(candidate)
        if len(candidate_prompts[2])<=budget:
            current,prepared=candidate,candidate_prompts;refilled=True
    if original_count>budget:
        fixed=len(prompts(dict(task,state=''))[2])
        query=task['question']['instructions']+' '+json.dumps(task['question'].get('criteria'),ensure_ascii=False)+' '+str(task.get('labels') or '')
        allowance=max(0,budget-fixed-limits['compression_reserve_tokens'])
        for _ in range(3):
            current=dict(task,state=shrink_state(source['state'],allowance,limits['truncation_head_frac'],
                backend=backend,query=query,head_tokens=limits['state_head_tokens'],
                tail_tokens=limits['state_tail_tokens'],chunk_tokens=limits['state_chunk_tokens']))
            prepared=prompts(current); size=len(prepared[2])
            if size<=budget: break
            allowance=max(0,allowance-(size-budget)-limits['compression_reserve_tokens'])
        # Compression cannot remove sacred rule/question/option material. The
        # final physical guard always bounds tensors, with explicit metadata.
    kept=len(prepared[2]); physical=limits['physical_max_prompt_tokens']
    hard_cut=kept>physical
    if hard_cut:
        def cut(ids):
            h=int(physical*limits['truncation_head_frac']);t=physical-h
            return ids if len(ids)<=physical else ids[:h]+ids[-t:]
        prepared=prepared[:2]+(cut(prepared[2]),cut(prepared[3]))
    return current,prepared,{'original_prompt_tokens':original_count if not precompressed else None,
        'original_prompt_tokens_estimate':math.ceil(len(raw)/limits['estimated_chars_per_token']) if precompressed else original_count,
        'long_input':long_input,'character_precompressed':precompressed,'adjacent_refill':refilled,
        'kept_prompt_tokens':len(prepared[2]),
        'slow_prompt_exceeds_budget':len(prepared[3])>min(budget,config['routing']['slow_max_prompt_tokens']),
        'truncated':current['state']!=task['state'] or hard_cut,
        'protected_content_exceeds_budget':kept>budget,'physical_truncated':hard_cut}


def sequence_scores(session, backend, groups, meter=None):
    """Full-vocabulary chain probabilities; copies branch KV, never prompt prefill.

    Keep every form (including repeated tokenizations) with its original weight.
    """
    cache = {(): session}
    continuation_tokens = 0
    normalized = {}
    path_scores = {(): 0.0}
    def score(ids):
        nonlocal continuation_tokens
        prefix = ()
        for token in ids:
            check_deadline(meter)
            child = prefix + (token,)
            if child not in path_scores:
                parent = cache[prefix]
                if prefix not in normalized:
                    wanted=sorted({encoded[len(prefix)] for forms in groups.values() for form in forms
                        if (encoded:=backend.encode(form))[:len(prefix)]==list(prefix) and len(encoded)>len(prefix)})
                    if hasattr(parent,'log_probs'):
                        normalized[prefix]=parent.log_probs(wanted)
                    else:
                        logits=parent.logits; lse=logsumexp(list(logits.values()))
                        normalized[prefix]={t:logits[t]-lse for t in wanted if t in logits}
                value=normalized[prefix].get(token)
                if value is None or not math.isfinite(value): raise RuntimeError('Missing candidate token')
                path_scores[child]=path_scores[prefix]+value
            prefix = child
            if prefix not in cache and len(prefix) < len(ids):
                branch = cache[prefix[:-1]].fork()
                if meter is not None and not getattr(branch, 'handles_meter', False):
                    meter['output'] += 1
                    meter['probes'] += 1
                    if getattr(branch,'meter',None) is None:
                        meter['engine_input']=meter.get('engine_input',0)+1; meter['engine_submitted']+=1
                branch.advance([token])
                continuation_tokens += 1
                cache[prefix] = branch
        return path_scores[prefix]
    digit_ids=getattr(backend,'digit_start_ids',None)
    if digit_ids is None:
        # Synthetic tokenizer fallback only; real backends scan the vocabulary.
        digit_ids={ids[0] for d in string.digits if len(ids:=backend.encode(d))==1}
    terminals={}
    def termination(ids):
        nonlocal continuation_tokens
        key=tuple(ids)
        if key not in terminals:
            if key not in cache:
                branch=cache[key[:-1]].fork()
                if meter is not None and not getattr(branch,'handles_meter',False):
                    meter['output']+=1;meter['probes']+=1
                    if getattr(branch,'meter',None) is None:
                        meter['engine_input']=meter.get('engine_input',0)+1;meter['engine_submitted']+=1
                branch.advance([key[-1]]);continuation_tokens+=1;cache[key]=branch
            se=cache[key]
            if hasattr(se,'non_digit_logprob'):
                value=se.non_digit_logprob(digit_ids)
            else:
                logits=se.logits; z=logsumexp(list(logits.values()))
                allowed=[v for t,v in logits.items() if t not in digit_ids]
                value=logsumexp(allowed)-z if allowed else -math.inf
            terminals[key]=value
        return terminals[key]
    result={}
    for label,forms in groups.items():
        is_prefix=label.isdigit() and any(other!=label and other.startswith(label) for other in groups)
        values=[]
        for form in forms:
            ids=backend.encode(form)
            if not ids: continue
            value=score(ids)
            if is_prefix and form.strip().isdigit(): value+=termination(ids)
            values.append(value)
        result[label]=logsumexp(values)
    return result, continuation_tokens


def slow_readout(backend, ids, groups, config, meter):
    """One autoregressive session: one prompt prefill, cached draft/marker/readout."""
    if hasattr(backend,'slow_session'):
        session,parsed=backend.slow_session(ids,config['routing']['unparsed_policy'],meter)
        if session is None: return None,meter['draft'],False,False,0
        # Marginalize textual whitespace forms; no full-vocabulary CPU copy.
        scores,probes=sequence_scores(session,backend,groups,meter)
        return scores,meter['draft']+meter['forced'],not parsed,False,probes
    meter['prompt'] += len(ids)
    meter['engine_submitted'] += len(ids)
    meter['engine_input']=meter.get('engine_input',0)+len(ids)
    meter['engine_calls']=meter.get('engine_calls',0)+1
    session = backend.open_session(ids)
    import random
    rng = random.Random(config['generation']['seed'])
    generated = []
    marker = config['prompt']['answer_marker']
    found = False
    for _ in range(config['generation']['max_new_tokens']):
        check_deadline(meter)
        logits = dict(session.logits)
        g = config['generation']
        for seen in set(generated):
            value = logits[seen]
            logits[seen] = value * g['repetition_penalty'] if value < 0 else value / g['repetition_penalty']
        token = max(logits, key=logits.get)
        if g['do_sample']:
            ordered = sorted(logits, key=logits.get, reverse=True)
            if g['top_k']:
                ordered = ordered[:g['top_k']]
            weights = [math.exp((logits[t]-logits[token])/g['temperature']) for t in ordered]
            mass, cutoff = 0, sum(weights)*g['top_p']
            end = 0
            while end < len(weights) and mass < cutoff:
                mass += weights[end]
                end += 1
            token = rng.choices(ordered[:end], weights=weights[:end])[0]
        generated.append(token)
        meter['output'] += 1
        meter['draft'] += 1
        meter['engine_generated']=meter.get('engine_generated',0)+1
        session.advance([token])
        text = backend.decode(generated)
        if re.search(r'(?m)^[ \t]*' + re.escape(marker) + r'$', text):
            found = True
            break
        if token in getattr(backend, 'eos_ids', ()):
            break
    if not found and config['routing']['unparsed_policy'] == 'fast':
        return None, len(generated), False, False, 0
    if not found:
        suffix = backend.encode('\n' + marker)
        meter['output'] += len(suffix)
        meter['forced'] += len(suffix)
        meter['engine_input']=meter.get('engine_input',0)+len(suffix);meter['engine_submitted']+=len(suffix)
        session.advance(suffix)
        generated.extend(suffix)
    # Whitespace is a form branch, never a top-1 destructive advance.
    whitespace = False
    scores, probes = sequence_scores(session, backend, groups, meter)
    return scores, len(generated), not found, whitespace, probes


def label_margin(labels, scores, kind="logratio"):
    """Aggregated label distribution at T=1, before any calibration/floor."""
    if len(labels) < 2:
        return math.inf
    values = sorted((scores[label] for label in labels), reverse=True)
    if kind == 'logratio':
        return values[0] - values[1]
    if kind != 'probdiff':
        raise ValueError('Invalid margin kind')
    weights = [math.exp(v-values[0]) for v in values]
    return (weights[0]-weights[1])/sum(weights)


def fast_readout(backend, ids, labels, groups, meter):
    meter["prompt"]+=len(ids)
    if not getattr(backend,"engine_metering",False):
        meter["engine_submitted"]+=len(ids);meter['engine_input']+=len(ids)
        meter['hf_prefill_tokens']+=len(ids)
    try:
        tokens = candidate_ids(backend, groups)
    except ValueError:
        tokens = {label: [] for label in labels}
    probes = 0
    if any(not values for values in tokens.values()):
        session=backend.open_session(ids,meter=meter) if getattr(backend,'engine_metering',False) else backend.open_session(ids)
        if hasattr(session,'cache'): session.meter=meter
        scores, probes = sequence_scores(session, backend, groups, meter)
    else:
        wanted = sorted({t for values in tokens.values() for t in values})
        logits=backend.next_logits(ids,wanted,meter=meter) if getattr(backend,'engine_metering',False) else backend.next_logits(ids,wanted)
        scores = aggregate(logits, tokens)
        if not getattr(backend,'engine_metering',False):
            meter['output']+=backend.readout_output_tokens;meter['probes']+=backend.readout_output_tokens
            meter['engine_generated']+=backend.readout_output_tokens
    return scores, probes + backend.readout_output_tokens


def release_failed_session(backend):
    import gc
    gc.collect()
    torch=getattr(backend,'torch',None)
    cuda=getattr(torch,'cuda',None)
    if cuda is not None:
        try: cuda.empty_cache()
        except Exception: pass


def fast_attempt_with_retry(backend,task,config,ids,labels,groups,meter):
    try:
        scores,out=fast_readout(backend,ids,labels,groups,meter)
        return scores,out,None
    except Exception as exc:
        if unrecoverable_engine_error(exc):
            raise
        cuda=getattr(getattr(backend,'torch',None),'cuda',None)
        oom_type=getattr(cuda,'OutOfMemoryError',MemoryError)
        if not isinstance(exc,(MemoryError,oom_type)) and 'out of memory' not in str(exc).lower():
            raise
    # Leave the failed forward traceback before clearing allocator state.
    release_failed_session(backend)
    import copy
    reduced=copy.deepcopy(config)
    budget=min(config['limits']['oom_retry_prompt_tokens'],max(1,config['limits']['max_prompt_tokens']//2))
    reduced['limits']['max_prompt_tokens']=budget
    reduced['limits']['physical_max_prompt_tokens']=budget
    _,(_,new_groups,new_ids,_),length=prepare_task(task,backend,reduced)
    scores,out=fast_readout(backend,new_ids,labels,new_groups,meter)
    return scores,out,length


def slow_attempt(backend,ids,groups,config,meter):
    # Catch in this frame, then leave except before collecting. The failed
    # readout/generate traceback (and its KV references) no longer exists.
    try:
        return slow_readout(backend,ids,groups,config,meter)
    except Exception as exc:
        if unrecoverable_engine_error(exc):
            raise ServiceFailure('unrecoverable engine failure') from exc
        timed_out=isinstance(exc,TimeoutError)
    release_failed_session(backend)
    if timed_out: raise TimeoutError('Slow deadline exceeded')
    raise RuntimeError('Slow session failed')


def bounded_slow_attempt(backend,ids,groups,config,meter,model_lock, *, on_service_failure=None):
    """Bound response wait; an in-flight synchronous engine call drains under lock.

    No worker may start another engine call after cancellation. Timeout usage is
    explicitly estimated and reconciliation rejects the run. An unrecoverable engine
    error is reported through on_service_failure even after the response timed out.
    """
    import queue,time
    result=queue.Queue(maxsize=1)
    timeout=config['generation'].get('wall_timeout_seconds',60)
    meter['cancel_event']=threading.Event();meter['deadline']=time.monotonic()+timeout
    def work():
        try:
            with model_lock:
                check_deadline(meter)
                value=slow_attempt(backend,ids,groups,config,meter)
                check_deadline(meter)
            result.put((True,value))
        except Exception as exc:
            fatal=unrecoverable_engine_error(exc)
            if fatal and on_service_failure is not None:
                on_service_failure()
            result.put((False,ServiceFailure if fatal else type(exc)))
    worker=threading.Thread(target=work,daemon=True);worker.start()
    try: ok,value=result.get(timeout=timeout)
    except queue.Empty:
        meter['cancel_event'].set();meter['engine_usage_unknown']=True
        raise TimeoutError('Slow wall timeout; in-flight call draining')
    if not ok:
        if value is ServiceFailure: raise ServiceFailure('unrecoverable engine failure')
        if value is TimeoutError: raise TimeoutError('Slow wall timeout')
        raise RuntimeError('Slow session failed')
    return value


def threshold_identity(config):
    import hashlib
    import inspect
    # The renderer source binds label layout, and settings bind both prompts.
    canonical=json.dumps({'prompt':{k:config['prompt'][k] for k in ('system','enable_thinking')},'render':inspect.getsource(render),'choice_codes':inspect.getsource(choice_codes),
        'generation':{k:v for k,v in config['generation'].items() if k not in ('max_new_tokens','reasoning_words','wall_timeout_seconds')}},sort_keys=True,ensure_ascii=False)
    m=config['model']
    return dict(model_key=m['model_key'],quant=m['quant'],adapter=m.get('adapter'),
                prompt_sha=hashlib.sha256(canonical.encode('utf-8')).hexdigest(),backend=m['backend'],
                margin_kind=config['routing']['margin_kind'],max_new_tokens=config['generation']['max_new_tokens'],
                reasoning_words=config['generation']['reasoning_words'],
                max_prompt_tokens=config['limits']['max_prompt_tokens'],request_contract='typesafe.criteria.v1')


def validate_threshold(config,allow_placeholder=False):
    identity=threshold_identity(config)
    fit=config['routing'].get('margin_threshold_fitted_for',{})
    reasons=[]
    if fit.get('placeholder',True): reasons.append('placeholder')
    if any(fit.get(k)!=identity[k] for k in ('model_key','quant','adapter','prompt_sha','margin_kind','max_prompt_tokens','request_contract')):
        reasons.append('identity mismatch')
    if fit.get('backend')!=identity['backend']: reasons.append('backend mismatch')
    if reasons and config['routing']['mode']=='routed' and not allow_placeholder:
        raise ValueError('Unfitted margin threshold: '+', '.join(reasons))
    return ['margin threshold degraded: '+', '.join(reasons)] if reasons else []


class Service:
    def __init__(self, config, calibration, backend, counter, *, allow_placeholder_threshold=False):
        validate_config(config)
        threshold_warnings=validate_threshold(config,allow_placeholder_threshold)
        self.model_lock = threading.RLock()
        self._failure_lock = threading.Lock()
        self.item_failure_streak = 0
        self.dead = False
        self.config, self.backend, self.counter = config, backend, counter
        counter.configure_cost(config)
        key = config['model']['model_key']
        self.calibration = {path: validate_calibration(calibration, key, path) for path in ('fast', 'slow')}
        model = calibration['models'][key]
        self.warnings = ['calibration placeholder: refit required'] if (
            model.get('placeholder') or any(b.get('placeholder') for b in self.calibration.values())) else []
        self.warnings.extend(threshold_warnings)
        if not config['routing'].get('cost_fuse',{}).get('enabled',True):
            self.warnings.append('cost fuse explicitly disabled')
        if 'fallback' in model:
            self.calibration['fallback']=validate_calibration(calibration,key,'fallback')
        else:
            self.calibration['fallback']=self.calibration['fast']
            self.warnings.append('fallback calibration missing: using fast block')

    def _mark_dead(self):
        with self._failure_lock:
            self.dead = True

    def is_dead(self):
        """Service failed: an unrecoverable engine error, the engine's own dead flag, or the
        third consecutive item failure. Reading the flag performs no engine call."""
        if self.dead:
            return True
        resource = self.backend
        for name in ('engine', 'llm_engine', 'engine_core', 'resources'):
            resource = getattr(resource, name, None)
        if getattr(resource, 'engine_dead', False) is True:
            self._mark_dead()
        return self.dead

    def _raise_if_service_failure(self, exc):
        if unrecoverable_engine_error(exc) or self.is_dead():
            self._mark_dead()
            raise ServiceFailure('unrecoverable engine failure') from exc

    def _item_failed(self):
        with self._failure_lock:
            self.item_failure_streak += 1
            if self.item_failure_streak >= ITEM_FAILURE_LIMIT:
                self.dead = True
            return self.item_failure_streak

    def _item_succeeded(self):
        with self._failure_lock:
            self.item_failure_streak = 0

    def health(self):
        dead = self.is_dead()
        warnings = list(self.warnings)
        if dead:
            warnings.append('service failure')
        reference=getattr(self.backend,'generation_reference',None)
        if reference and not reference['verified']:warnings.append('generation reference '+reference['status'])
        if hasattr(self.backend,'forward_counters') and not getattr(self.backend,'stats_available',False):warnings.append('HF forward hook unavailable; usage estimated')
        if getattr(self.backend,'forward_counters',{}).get('uncounted_forwards',0):warnings.append('HF uncounted forwards; usage estimated')
        if getattr(self.counter, 'memory_only', False):
            warnings.append('quota persistence unavailable: memory counts, fast only')
        return {'status': 'error' if dead else 'warning' if warnings else 'ok', 'warnings': warnings,
                'info': {'intended_overrides': (reference or {}).get('intended_overrides',{})},
                'model_key': self.config['model']['model_key'], 'mode':self.config['routing']['mode'],
                'quota_tier':self.config['routing'].get('tier','normal'), 'budget_tiers':self.config['routing'].get('tiers',{}),
                'threshold_identity':threshold_identity(self.config),
                'cost_fuse':self.counter.cost_fuse.snapshot(self.config['routing']['quota']),
                'accuracy_only':self.config.get('bench',{}).get('accuracy_only',False),
                'generation_reference':({k:v for k,v in reference.items() if k!='intended_overrides'} if reference else reference),'quantization_kernel':getattr(self.backend,'quantization_kernel',None),
                'prefix_caching':getattr(self.backend,'prefix_caching',None),
                'supports_slow_session':callable(getattr(self.backend,'slow_session',None)) or callable(getattr(self.backend,'open_session',None))}

    def engine_stats(self):
        with self.model_lock:
            return self.backend.engine_stats()

    def warmup(self):
        ids = self.backend.prompt_ids(self.config['prompt']['system'], 'Ready?', False)
        self.backend.next_logits(ids, [self.backend.encode('yes')[0]])
        self.backend.generate(ids, 1)
        meter=new_meter()
        with self.model_lock:
            if callable(getattr(self.backend,'slow_session',None)):
                session,_=self.backend.slow_session(ids,'force_answer',meter,
                    max_new_tokens=self.config['server']['warmup_slow_tokens'])
                if session is not None:
                    branch=session.fork();branch.advance(self.backend.encode('yes'))
                    if hasattr(branch,'log_probs'): branch.log_probs([self.backend.encode('yes')[0]])
                    else: _=branch.logits
                    del branch,session
            elif callable(getattr(self.backend,'open_session',None)):
                import copy
                warm=copy.deepcopy(self.config)
                warm['generation']['max_new_tokens']=self.config['server']['warmup_slow_tokens']
                warm['routing']['unparsed_policy']='force_answer'
                slow_attempt(self.backend,ids,{'yes':['yes',' yes']},warm,meter)

    def _one(self, task):
        with self.counter.accounting():
            if self.is_dead():
                raise ServiceFailure('service failure')
            try:
                answer,detail=self._one_impl(task)
                if self.is_dead():
                    raise ServiceFailure('service failure')
                detail['threshold_identity']=threshold_identity(self.config)
                detail['cost_fuse']=self.counter.record_cost(detail['usage_bases'][detail['accounting']],detail['path']=='slow')
                try:
                    json.dumps([answer,detail],allow_nan=False)
                except ValueError as exc:
                    raise ItemFailure('non-finite value in response') from exc
            except ItemFailure as exc:
                if self._item_failed() >= ITEM_FAILURE_LIMIT:
                    raise ServiceFailure('consecutive item failures') from exc
                raise
            except ServiceFailure:
                self._mark_dead()
                raise
            self._item_succeeded()
            return answer,detail

    def _one_impl(self, task):
        c, backend = self.config, self.backend
        uncounted_before=getattr(backend,'forward_counters',{}).get('uncounted_forwards',0)
        labels, _, groups = render(task)
        qtype = task['question']['type']
        detail = {'path':'fast', 'route_reason':'error', 'router_score':0.0,
                  'prompt_tokens':0, 'completion_tokens':0, 'n_new_tokens':0,
                  'prompt_tokens_details':{'cached_tokens':0,'engine_submitted_tokens':0},
                  'fast_scores':None,'slow_scores':None,'budget_tier':'normal',
                  'readout_output_tokens':0, 'forced_answer':False, 'draft_prediction':None}
        warning = None
        meters = []
        try:
            kept_task, (_, groups, fast_ids, slow_ids), length = prepare_task(task, backend, c)
            detail.update(length)
            long_item=length.get('long_input',False) or c['routing'].get('tier')=='long' or length.get('slow_prompt_exceeds_budget',False) or any(length[k] for k in ('truncated','character_precompressed','protected_content_exceeds_budget','physical_truncated'))
            hard=long_item if c['routing'].get('long_item_slow_policy','fast_only')=='fast_only' else (
                length['character_precompressed'] or (length['original_prompt_tokens'] or 0)>c['routing']['slow_max_prompt_tokens'] or length['protected_content_exceeds_budget'])
            hard=hard or c['routing'].get('tier')=='long' or length.get('slow_prompt_exceeds_budget',False)
            detail['long_item']=long_item
            r = c['routing']
            fast_scores = None
            fast_meter=new_meter()
            meters.append(fast_meter)
            try:
                with self.model_lock:
                    fast_scores, out, retry_length = fast_attempt_with_retry(backend,task,c,fast_ids,labels,groups,fast_meter)
            except Exception as exc:
                self._raise_if_service_failure(exc)
                raise ItemFailure('item could not be processed') from exc
            if retry_length is not None:
                detail.update(retry_length)
                detail['fast_oom_retry']=True
                hard=True
            detail['prompt_tokens']+=fast_meter['prompt']
            detail['completion_tokens']+=fast_meter['output']
            detail['readout_output_tokens']+=fast_meter['probes']
            detail['prompt_tokens_details'].update(cached_tokens=fast_meter['cached'],engine_submitted_tokens=fast_meter['engine_submitted'])
            if fast_meter.get('engine_usage_unknown'): detail['usage_estimated']=True
            if fast_meter.get('cached_unknown'): detail['prompt_tokens_details']['cached_tokens_known']=False
            detail['fast_prompt_tokens']=fast_meter['prompt']
            detail['raw_label_scores']=detail['fast_scores']=fast_scores
            margin = label_margin(labels, fast_scores, r['margin_kind']) if fast_scores is not None else None
            detail.update(fast_margin=margin if margin is None or math.isfinite(margin) else None,
                          fast_margin_infinite=margin == math.inf, margin_kind=r['margin_kind'])
            wants_slow = fast_scores is not None and not hard and (
                r['mode'] == 'uniform' or (r['mode'] == 'routed' and margin < r['margin_threshold']))
            wants_slow = wants_slow and (callable(getattr(backend,'slow_session',None)) or callable(getattr(backend,'open_session',None)))
            detail['budget_tier']='long' if hard else 'hard' if wants_slow else 'normal'
            detail['admission_quota'],detail['admission_quota_source']=self.counter.cost_fuse.quota(r['quota'])
            with self.counter.decision(wants_slow, quota=detail['admission_quota']) as decision:
                detail.update(path=decision.path, route_reason='oom_retry' if detail.get('fast_oom_retry') else 'length' if hard else decision.reason,
                              quota_at_reservation=decision.snapshot)
                scores, block = fast_scores, 'fast'
                if decision.path == 'slow':
                    meter = new_meter()
                    meters.append(meter)
                    try:
                        scores, count, forced, whitespace, probes = bounded_slow_attempt(backend,slow_ids,groups,c,meter,self.model_lock,on_service_failure=self._mark_dead)
                        detail['slow_raw_label_scores']=detail['slow_scores']=scores
                        block = 'slow' if scores is not None else 'fallback'
                        detail['route'] = ('slow_forced_answer' if forced else 'slow') if scores is not None else 'slow_unparsed_fast'
                        if scores is None:
                            scores = fast_scores
                        detail.update(n_new_tokens=count, forced_answer=forced, whitespace_step=whitespace)
                    except Exception as exc:
                        self._raise_if_service_failure(exc)
                        scores, block = fast_scores, 'fallback'
                        warning = 'slow_failed: fast answer kept'
                        detail.update(fallback='slow_to_fast', route='slow_timeout_fast' if isinstance(exc,TimeoutError) else 'slow_failed_fast')
                        if isinstance(exc,TimeoutError):
                            warning='slow_timeout: fast answer kept'
                            detail['usage_estimated']=True
                            # Detach response accounting from the draining worker.
                            meter=dict(meter);meters[-1]=meter
                    finally:
                        if meter.get('engine_usage_unknown'): detail['usage_estimated']=True
                        if meter.get('cached_unknown'): detail['prompt_tokens_details']['cached_tokens_known']=False
                        detail['prompt_tokens'] += meter['prompt']
                        detail['completion_tokens'] += meter['output']
                        detail['n_new_tokens'] = meter['draft'] + meter['forced']
                        detail['readout_output_tokens'] += meter['probes']
                        detail['prompt_tokens_details']['cached_tokens'] += meter['cached']
                        detail['prompt_tokens_details']['engine_submitted_tokens'] += meter['engine_submitted']
                detail['calibration_block']=block
                if scores is None:
                    raise ItemFailure('no label scores')
                probs = probabilities(labels, scores, qtype, self.calibration.get(block, self.calibration['fast']))
            detail['quota_after_completion'] = decision.snapshot
        except (ItemFailure, ServiceFailure, QuotaStateError):
            raise
        except Exception as exc:
            # No substitute answer: an engine failure is service-level (HTTP 503); any other
            # failure fails this item (HTTP 422), which the harness scores as wrong.
            self._raise_if_service_failure(exc)
            raise ItemFailure('item could not be processed') from exc
        if (getattr(backend,'forward_counters',{}).get('uncounted_forwards',0)>uncounted_before
                or hasattr(backend,'forward_counters') and not getattr(backend,'stats_available',False)):
            detail['usage_estimated']=True
            warning='HF count unavailable: usage estimated'
        detail['backend_warnings']=[w for m in meters for w in m.get('warnings',[])]
        if any(m.get('cached_unknown') for m in meters):detail['usage_estimated']=True
        submitted=sum(m['engine_submitted'] for m in meters)
        cached=sum(m['cached'] for m in meters)
        generated=sum(m.get('engine_generated',m['output']) for m in meters)
        engine_input=sum(m.get('engine_input',m['engine_submitted']-m['cached']) for m in meters)
        if detail.get('usage_estimated'):
            engine_input=max(engine_input,submitted)+c['generation']['max_new_tokens']
            generated+=c['generation']['max_new_tokens']
        detail['input_tokens_details']=dict(cached_tokens=cached,engine_submitted_tokens=submitted,
            engine_calls=sum(m.get('engine_calls',0) for m in meters),engine_generated_tokens=generated,
            cached_tokens_known=not any(m.get('cached_unknown') for m in meters),
            generate_nonempty_calls=sum(m.get('generate_nonempty_calls',0) for m in meters),
            hf_prefill_tokens=sum(m.get('hf_prefill_tokens',0) for m in meters))
        detail['usage_bases']={
            'engine':dict(input_tokens=engine_input,output_tokens=generated),
            'each_once':dict(input_tokens=detail['prompt_tokens'],output_tokens=detail['completion_tokens']),
            'submitted':dict(input_tokens=submitted,output_tokens=detail['completion_tokens'])}
        if detail.get('usage_estimated'):
            for values in detail['usage_bases'].values():
                values['input_tokens']=max(values['input_tokens'],engine_input)
                values['output_tokens']=max(values['output_tokens'],generated)
        detail['accounting']=c.get('usage',{}).get('basis','engine')
        detail['kept_prompt_tokens_max'] = detail.get('kept_prompt_tokens', 0)
        detail['total_tokens'] = detail['prompt_tokens'] + detail['completion_tokens']
        detail['prediction'] = max(labels, key=probs.get)
        request_type = task['question']['request_type']
        if qtype == 'noul':
            style = c.get('interface', {}).get('boolean_keys', 'request')
            if style == 'true_false' or (style == 'request' and request_type == 'boolean'):
                probs = {'false':probs['no'], 'true':probs['yes']}
        answer = {'type':request_type, 'probabilities':probs}
        if qtype == 'noul': answer['noul'] = float(probs.get('yes', probs.get('true')))
        elif qtype == 'choice': answer['choice'] = detail['prediction']
        if warning:
            answer['warning'] = warning
        return answer, detail

    def answer(self, payload):
        if self.is_dead():
            raise ServiceFailure('service failure')
        tasks = normalize_request(payload, self.config['server']['max_questions'])
        answers, details, errors, noul, choice = {}, {}, {}, {}, {}
        for name, task in tasks:
            answer, detail = self._one(task)
            answers[name], details[name] = answer, detail
            if 'warning' in answer:
                errors[name] = answer['warning']
            if task['question']['type'] == 'noul':
                noul[name] = answer['probabilities'].get('true', answer['probabilities'].get('yes'))
            else:
                choice[name] = detail['prediction']
        basis=self.config.get('usage',{}).get('basis','engine')
        alternatives={b:{k:sum(d['usage_bases'][b][k] for d in details.values())
            for k in ('input_tokens','output_tokens')} for b in ('engine','each_once','submitted')}
        p,o=alternatives[basis]['input_tokens'],alternatives[basis]['output_tokens']
        token_details={k:sum(d['input_tokens_details'][k] for d in details.values()) for k in
            ('cached_tokens','engine_submitted_tokens','engine_calls','engine_generated_tokens','generate_nonempty_calls','hf_prefill_tokens')}
        token_details['cached_tokens_known']=all(d['input_tokens_details']['cached_tokens_known'] for d in details.values())
        result={'answers':answers,'model':self.config['model']['name'],
            'usage':dict(input_tokens=p,output_tokens=o,prompt_tokens=p,completion_tokens=o,total_tokens=p+o,
                cost_fuse=next(reversed(details.values()))['cost_fuse'],
                accounting=basis,usage_estimated=any(d.get('usage_estimated',False) for d in details.values()),
                alternatives=alternatives,questions=details,input_tokens_details=token_details,
                prompt_tokens_details=token_details,token_accounting=basis)}
        for key, values in (('noul',noul),('choice',choice)):
            if values:
                result[key] = next(iter(values.values())) if len(tasks) == 1 else values
        if errors:
            result['partial_errors'] = errors
        return result