"""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'], }