"""Server-owned evaluation configuration; private resources never reach clients.""" from __future__ import annotations import hashlib import json import math import os from datetime import datetime, timezone from pathlib import Path from threading import Lock from urllib.parse import urlparse from uuid import uuid4 from src.evaluation.scoring import SubmissionError, evaluate, load_labels, summary, validate_predictions from src.evaluation.storage import StorageError, SubmissionStore MAX_UPLOAD_BYTES = 2 * 1024 * 1024 DEFAULT_REPO = 'XLearning-SCU/ARK-Bench-Evaluation' class EvaluationUnavailable(RuntimeError): pass class EvaluationService: def __init__(self, root: Path): self.root = root self.manifest = json.loads((root / 'evaluation_manifest.json').read_text()) self.token = os.environ.get('ARK_EVALUATION_TOKEN') self.hosted = bool(os.environ.get('SPACE_ID')) # Setting a token explicitly selects durable private Hub storage locally too. self.remote = self.hosted or bool(self.token) self.repo = os.environ.get('ARK_EVALUATION_REPO', DEFAULT_REPO) self.labels_revision = os.environ.get('ARK_LABELS_REVISION', 'main') self.labels_path = root / 'evaluation' / 'evaluation_labels.jsonl' self.knowledge_path = root / 'evaluation' / 'knowledge_domains.json' self.cache_dir = root / 'evaluation-cache' / 'private' self.store = (SubmissionStore(root / 'submissions', self.repo if self.remote else None, self.token) if not self.remote or self.token else None) self._labels = None self._labels_digest = None self._knowledge_digest = None self._resolved_revision = None self._lock = Lock() def labels(self): with self._lock: if self._labels is not None: return self._labels try: if self.remote: if not self.token: raise EvaluationUnavailable('Online evaluation is awaiting maintainer configuration.') from huggingface_hub import HfApi, hf_hub_download info = HfApi(token=self.token).dataset_info(self.repo, revision=self.labels_revision) if not info.private: raise EvaluationUnavailable('Private evaluation storage is not configured correctly.') self._resolved_revision = info.sha path = Path(hf_hub_download( self.repo, 'evaluation_labels.jsonl', repo_type='dataset', revision=info.sha, token=self.token, cache_dir=self.cache_dir, )) knowledge_path = Path(hf_hub_download( self.repo, 'knowledge_domains.json', repo_type='dataset', revision=info.sha, token=self.token, cache_dir=self.cache_dir, )) else: path = self.labels_path knowledge_path = self.knowledge_path labels = load_labels(path, self.manifest, knowledge_path) self._labels_digest = hashlib.sha256(path.read_bytes()).hexdigest() self._knowledge_digest = hashlib.sha256(knowledge_path.read_bytes()).hexdigest() self._labels = labels return labels except EvaluationUnavailable: raise except Exception: raise EvaluationUnavailable('Evaluation is temporarily unavailable. Please try again later.') from None def protocol(self): available = True message = 'Scores are returned after private storage succeeds; publication requires maintainer review.' try: self.labels() if not self.remote: message = 'Local preview: submissions are saved on this computer for maintainer review.' except EvaluationUnavailable as exc: available, message = False, str(exc) return { 'benchmark_version': self.manifest['benchmark_version'], 'dataset_revision': self.manifest['dataset_revision'], 'dataset_repo': self.manifest['dataset_repo'], 'query_count': len(self.manifest['queries']), 'top_k': self.manifest['top_k'], 'max_upload_bytes': MAX_UPLOAD_BYTES, 'available': available, 'message': message, 'template_url': '/api/evaluation/template', } def template(self): # Generated only from the public manifest; no answer-derived ranking examples. return ''.join(json.dumps({'topic': r['topic'], 'query_id': r['query_id'], 'retrieved_ids': []}) + '\n' for r in self.manifest['queries']) def _evaluate(self, content: bytes, fields: dict[str, str]): if fields.get('benchmark_version') != self.manifest['benchmark_version']: raise SubmissionError('Benchmark version mismatch. Reload the page and use the current template.') if fields.get('dataset_revision') != self.manifest['dataset_revision']: raise SubmissionError('Dataset revision mismatch. Use the dataset revision shown in the submission protocol.') model = model_metadata(fields) predictions = validate_predictions(content, self.manifest) score = evaluate(predictions, self.labels(), self.manifest, model) return score def evaluate(self, content: bytes, fields: dict[str, str]): score = self._evaluate(content, fields) evaluation_id = uuid4().hex return { 'evaluation_id': evaluation_id, 'status': 'evaluated', 'score': score, 'summary': summary(score), 'message': 'Evaluation complete. Review the aggregate scores, then submit the report for maintainer review.', } def submit_report(self, content: bytes, fields: dict[str, str]): evaluation_id = fields.get('evaluation_id') if not isinstance(evaluation_id, str) or len(evaluation_id) != 32 \ or any(character not in '0123456789abcdef' for character in evaluation_id): raise SubmissionError('Invalid evaluation receipt. Evaluate the file again before submitting.') score = self._evaluate(content, fields) record = { 'submission_id': evaluation_id, 'status': 'evaluated', 'created_at': datetime.now(timezone.utc).isoformat(), 'submission_sha256': hashlib.sha256(content).hexdigest(), 'labels_sha256': self._labels_digest, 'labels_revision': self._resolved_revision, 'knowledge_domains_sha256': self._knowledge_digest, 'manifest_sha256': hashlib.sha256((self.root / 'evaluation_manifest.json').read_bytes()).hexdigest(), 'score': score, } try: self.store.save(record, content) except StorageError: # A retry after an interrupted response reuses the same evaluated record. try: existing = self.store.load(evaluation_id) existing_content = self.store.read_predictions(evaluation_id) same = ( existing.get('status') in ('evaluated', 'pending_review') and existing_content == content and existing.get('submission_sha256') == record['submission_sha256'] and existing.get('labels_sha256') == record['labels_sha256'] and existing.get('knowledge_domains_sha256') == record['knowledge_domains_sha256'] and existing.get('manifest_sha256') == record['manifest_sha256'] and existing.get('score', {}).get('model') == score['model'] and existing.get('score', {}).get('results') == score['results'] ) if not same: raise ValueError except Exception: raise EvaluationUnavailable( 'Could not save this submission. Please retry later.') from None try: self.store.submit_report(evaluation_id) except StorageError: raise EvaluationUnavailable( 'Could not submit this report for review. Please retry later.') from None message = ('Report submitted. A maintainer has been notified for review.' if self.remote else 'Report submitted locally for maintainer review.') return { 'status': 'pending_review', 'message': message, } def model_metadata(fields): name = fields.get('model_name', '').strip() if not name or len(name) > 120 or any(ord(c) < 32 for c in name): raise SubmissionError('Model name must contain 1–120 printable characters.') url = fields.get('model_url', '').strip() if url: try: parsed = urlparse(url) if (len(url) > 500 or parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.username or parsed.password or any(c.isspace() for c in url)): raise ValueError except ValueError: raise SubmissionError('Model URL must be a valid HTTP(S) URL without credentials.') from None params = fields.get('params_b', '').strip() try: params = float(params) if params else None if params is not None and (not math.isfinite(params) or params < 0): raise ValueError except ValueError: raise SubmissionError('Parameters must be a finite non-negative number in billions.') from None return {'name': name, 'repo_name': None, 'repo_link': url or None, 'params_b': params, 'model_type': 'multimodal_embedding'}