systemone_api.py

"""Public SystemOne transport compatibility; native model probabilities are unchanged."""
from copy import deepcopy
from datetime import date
import json
from pathlib import Path
from fastapi import HTTPException
from model_registry import MODEL_ORDER, PROFILES

# Descending published parameter size, with general Kai before specialist Lex at 0.6B.
ORDER = MODEL_ORDER
MODEL_RELEASES_PATH = Path(__file__).with_name('MODEL_RELEASES.json')
CONFIDENCE_DEFINITION = "max(p); native peak probability, not calibrated correctness or TypeSafe formula equivalence"


def arrange_models(registry):
    return {key: registry[key] for key in ORDER if key in registry}


def resolve_model(value, registry):
    if not isinstance(value, str) or not value.strip():
        raise HTTPException(422, "Specify an available Decision model")
    value = value.casefold()
    for key, item in registry.items():
        aliases = [key, item['label'], item.get('repo_id', '')]
        if item.get('repo_id'):
            aliases.append(item['repo_id'].rsplit('/', 1)[-1])
        if value in (alias.casefold() for alias in aliases if alias):
            return key
    raise HTTPException(422, "This model is not available in this Studio")


def resolve_canonical_model(value, registry):
    """Resolve the public SystemOne identifier without accepting wire aliases."""
    if not isinstance(value, str) or not value.strip():
        raise HTTPException(422, "Specify an available canonical Decision model")
    for key, item in registry.items():
        if value == item['repo_id']:
            return key
    raise HTTPException(
        422,
        "Use the exact Hugging Face repository ID for an available Decision model.",
    )


def _release_date(value):
    if not isinstance(value, str):
        raise ValueError('Release date must be an ISO calendar date')
    try:
        parsed = date.fromisoformat(value)
    except ValueError as exc:
        raise ValueError('Release date must be an ISO calendar date') from exc
    if parsed.isoformat() != value:
        raise ValueError('Release date must be an ISO calendar date')
    return value


def load_model_releases(path=None):
    """Validate presentation metadata; never use it to admit a serving artifact."""
    source = MODEL_RELEASES_PATH if path is None else Path(path)
    document = json.loads(source.read_text(encoding='utf-8'))
    if not isinstance(document, dict) or set(document) != {'release', 'models'}:
        raise ValueError('Invalid model release metadata')
    default_date = _release_date(document['release'])
    entries = document['models']
    if not isinstance(entries, list) or not 1 <= len(entries) <= len(ORDER):
        raise ValueError('Invalid model release entries')
    releases = {}
    for entry in entries:
        if not isinstance(entry, dict) or set(entry) - {
            'label', 'repo_id', 'revision', 'manifest_sha256',
            'complete_input_tokens', 'release_date',
        } or {'repo_id', 'revision', 'manifest_sha256'} - set(entry):
            raise ValueError('Invalid model release entry')
        repo_id = entry['repo_id']
        revision = entry['revision']
        manifest = entry['manifest_sha256']
        if (not isinstance(repo_id, str) or repo_id not in {
                profile['repo_id'] for profile in PROFILES.values()
            } or repo_id in releases
            or not isinstance(revision, str) or len(revision) != 40
            or any(c not in '0123456789abcdef' for c in revision)
            or not isinstance(manifest, str) or len(manifest) != 64
            or any(c not in '0123456789abcdef' for c in manifest)):
            raise ValueError('Invalid model release identity')
        if 'label' in entry and (not isinstance(entry['label'], str) or not entry['label']):
            raise ValueError('Invalid model release label')
        if 'complete_input_tokens' in entry and (
            type(entry['complete_input_tokens']) is not int
            or entry['complete_input_tokens'] < 1
        ):
            raise ValueError('Invalid model release input limit')
        releases[repo_id] = {
            'revision': revision,
            'manifest_sha256': manifest,
            'release_date': _release_date(entry.get('release_date', default_date)),
        }
    return releases


def public_models(registry, *, releases=None):
    if releases is None:
        try:
            releases = load_model_releases()
        except (OSError, ValueError, UnicodeError, RecursionError):
            # Release dates are presentation metadata, not model admission.
            # Explicit load_model_releases() callers still get validation errors.
            releases = {}
    result = []
    for key, item in registry.items():
        public = dict(
            item, id=item['repo_id'], wire_id=key, name=item['label'],
            description=item.get('description', 'Decision foundation model'),
        )
        release = releases.get(item['repo_id'])
        if (release is not None and item.get('revision') == release['revision']
                and item['manifest_sha256'] == release['manifest_sha256']):
            public['release_date'] = release['release_date']
        result.append(public)
    return result


def sdk_response(result, *, public_model=None):
    """Add the SDK's required distribution statistic without altering native outputs."""
    result = deepcopy(result)
    if public_model is not None:
        result['model'] = public_model
    groups = [result['answers']] if 'answers' in result else [r['answers'] for r in result['results']]
    for answers in groups:
        for answer in answers.values():
            if answer['type'] in ('choice', 'score'):
                probabilities = list(answer['probabilities'].values())
                answer['confidence'] = max(probabilities)
    result['profile'] = dict(result.get('profile', {}), confidence_definition=CONFIDENCE_DEFINITION)
    return result


def public_response(result, *, public_model, batch):
    """Project native diagnostics onto the strict public Decision envelope."""
    prepared = sdk_response(result, public_model=public_model)

    def answers_only(answers):
        fields = {
            'noul': ('type', 'noul'),
            'choice': ('type', 'choice', 'confidence', 'probabilities'),
            'score': ('type', 'score', 'confidence', 'legend', 'probabilities'),
        }
        return {
            key: {field: answer[field] for field in fields[answer['type']]}
            for key, answer in answers.items()
        }

    if batch:
        return {
            'model': public_model,
            'results': [
                {'id': row['id'], 'answers': answers_only(row['answers']), 'usage': row['usage']}
                for row in prepared['results']
            ],
            'usage': prepared['usage'],
        }
    return {
        'model': public_model,
        'answers': answers_only(prepared['answers']),
        'usage': prepared['usage'],
    }