From 75c034d58856dd939eec6278e3e889272d711924 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Sun, 20 Sep 2026 10:47:22 +0800 Subject: [PATCH] fix: recover eligible model GPU routes --- README.md | 17 +- modelhub_submmit_api/README.md | 11 +- modelhub_submmit_api/hf_discovery.py | 30 +++- modelhub_submmit_api/main.py | 119 ++++++++++++- modelhub_submmit_api/market_intelligence.py | 14 ++ modelhub_submmit_api/modelhub_client.py | 39 ++++- modelhub_submmit_api/poll_runner.py | 11 +- modelhub_submmit_api/version.py | 2 +- tests/test_concurrency.py | 3 +- tests/test_market_intelligence.py | 8 + tests/test_modelscope_discovery.py | 25 +++ tests/test_poll_policy.py | 4 + tests/test_submission_safety.py | 182 +++++++++++++++++++- 13 files changed, 436 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index dd6de194ec..41a22dc437 100644 --- a/README.md +++ b/README.md @@ -451,11 +451,24 @@ routes per model, preserving successful-route and queue ordering without discarding capacity on lower-ranked GPUs when the preferred tranche cannot run the model. +Version `2026.09.20.1` corrects the community verification precheck: an exact +model/GPU route marked `未验证` with no records is eligible, while `已验证`, `验证中`, +and unknown states stay blocked. A `验证失败` route may use a different framework +only when every previous record explicitly failed and names its framework; the +fresh check before submission rejects any framework that has already failed on +that GPU. The official framework evidence and deterministic preflight gates +still apply. When recent-update discovery is exhausted, a bounded 500-model +download-ranked ModelScope scan checks a rotating 200-row window of older +popular models without exceeding the platform's 3,000-row pagination ceiling. +Architecture history backfill now +sends the date format accepted by ModelHub. The full-repository-size OOM rule is +unchanged because ModelHub's own preflight uses that size. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-v34 -m "ModelHub agent 2026.09.18.1" -git push origin main agent-v34 +git tag -a agent-v35 -m "ModelHub agent 2026.09.20.1" +git push origin main agent-v35 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 833ae5c1ac..55ea2eb683 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -115,14 +115,19 @@ bash run_poll.sh --dry-run - Strategy state is stored in `.modelhub_state/gpu_strategy.json`; a generation never recalculates during candidate submission. - Candidate discovery starts with the configured recent window, then automatically expands to - 7 days, 30 days, and older history (up to 3,000 models) when the recent pool is exhausted. + 7 days, 30 days, and the first 3,000 raw ModelScope rows ordered by update time. If the + candidate pool is still short, it rotates through up to 200 download-ranked rows per cycle + as an older-model fallback. - Model verification responses are cached across poll cycles for 15 minutes. Local model/GPU failures cool down after 24 hours instead of remaining permanently blocked. - Community deduplication is model/GPU-specific: another GPU's adaptation does not block the - current GPU. Every actual submission performs a fresh uncached check for its exact GPU. + current GPU. `未验证` with no records is eligible. `验证失败` is eligible only with a + different publicly vetted framework after all recorded attempts explicitly failed; + `已验证`, `验证中`, and ambiguous records remain blocked. Every actual submission + performs a fresh uncached check for its exact GPU and framework. - If the community lookup is unavailable, submission is deferred. A platform model-uniqueness rejection permanently excludes only that model/GPU combination from future local retries. -- Each model can be submitted at most once per GPU. +- The same framework is not retried for a model/GPU after a recorded failure. - Multiple ModelHub tokens are pooled and used to route submissions to the account with available async capacity. - Concurrent submissions reserve account slots locally, and an account-capacity race automatically falls through to another account. - Concurrent local processes claim model/GPU pairs in `.modelhub_state/submission_claims.jsonl`; shared ledger, history, and outcome files use process locks and atomic replacement. diff --git a/modelhub_submmit_api/hf_discovery.py b/modelhub_submmit_api/hf_discovery.py index 87c07de6f5..5cb9bf49b9 100644 --- a/modelhub_submmit_api/hf_discovery.py +++ b/modelhub_submmit_api/hf_discovery.py @@ -86,7 +86,7 @@ class HuggingFaceDiscovery: self._model_last_modified_cache: OrderedDict[str, datetime | None] = OrderedDict() self._model_last_modified_lock = threading.Lock() self._model_page_cache: OrderedDict[ - tuple[str, int, int], tuple[float, list[dict[str, Any]]] + tuple[str, str, int, int], tuple[float, list[dict[str, Any]]] ] = OrderedDict() self._model_page_cache_ttl = max( 0.0, @@ -108,6 +108,8 @@ class HuggingFaceDiscovery: self._unsupported_task_filters: dict[str, float] = {} self._model_card_cache: OrderedDict[str, dict[str, Any]] = OrderedDict() self._model_card_lock = threading.Lock() + self._popular_page_lock = threading.Lock() + self._popular_page_cursor = 0 @staticmethod def _remember_lru(cache: OrderedDict, key: Any, value: Any, limit: int) -> None: @@ -124,6 +126,8 @@ class HuggingFaceDiscovery: min_downloads: int, updated_after=None, read_concurrency: int = 1, + sort: str = "last_modified", + rotate_pages: bool = False, ) -> list[HFModelSummary]: if not pipeline_tags: return [] @@ -132,6 +136,17 @@ class HuggingFaceDiscovery: del read_concurrency per_tag_limit = limit if len(pipeline_tags) <= 1 else max(10, (limit + len(pipeline_tags) - 1) // len(pipeline_tags)) + start_page = 1 + if rotate_pages and sort == "downloads": + window_pages = max(1, min(60, (per_tag_limit + 49) // 50)) + window_count = max(1, 60 // window_pages) + with self._popular_page_lock: + start_page = 1 + (self._popular_page_cursor % window_count) * window_pages + self._popular_page_cursor += 1 + print( + f"[modelscope] popular_window=start_page:{start_page} pages:{window_pages}", + flush=True, + ) for pipeline_tag in pipeline_tags: for model in self._query_recent_models( @@ -139,6 +154,8 @@ class HuggingFaceDiscovery: limit=per_tag_limit, min_downloads=min_downloads, updated_after=updated_after, + sort=sort, + start_page=start_page, ): current = deduped.get(model.repo_id) if current is None or (model.last_modified or parse_datetime("1970-01-01")) > ( @@ -156,6 +173,8 @@ class HuggingFaceDiscovery: limit: int, min_downloads: int, updated_after=None, + sort: str = "last_modified", + start_page: int = 1, ) -> list[HFModelSummary]: # ModelScope OpenAPI rejects values above 50 with InputParameterError. page_size = min(max(1, limit), 50) @@ -163,8 +182,9 @@ class HuggingFaceDiscovery: task_tag = MODELSCOPE_TASK_TAGS.get(pipeline_tag, pipeline_tag) filter_disabled = self._unsupported_task_filters.get(task_tag, 0.0) > time.monotonic() models: list[HFModelSummary] = [] - for page_number in range(1, (max_items + page_size - 1) // page_size + 1): - cache_key = ("*" if filter_disabled else task_tag, page_number, page_size) + page_count = (max_items + page_size - 1) // page_size + for page_number in range(start_page, min(61, start_page + page_count)): + cache_key = (sort, "*" if filter_disabled else task_tag, page_number, page_size) cached = self._model_page_cache.get(cache_key) if cached is not None and time.monotonic() - cached[0] < self._model_page_cache_ttl: self._model_page_cache.move_to_end(cache_key) @@ -177,7 +197,7 @@ class HuggingFaceDiscovery: query = { "page_number": page_number, "page_size": page_size, - "sort": "last_modified", + "sort": sort, } if not filter_disabled: query["filter.task"] = task_tag @@ -201,6 +221,8 @@ class HuggingFaceDiscovery: limit=limit, min_downloads=min_downloads, updated_after=updated_after, + sort=sort, + start_page=start_page, ) print( f"[modelscope] list_models_error task={task_tag} page={page_number} " diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 3c5189061d..89386ce858 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -10,7 +10,7 @@ from pathlib import Path from typing import Any from common import parse_datetime, runtime_instance_id, utc_now, write_json, write_jsonl -from candidate_preflight import CandidatePreflightAdvisor +from candidate_preflight import CandidatePreflightAdvisor, MODEL_LOAD_OVERHEAD from config_optimizer import SafeConfigOptimizer from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH from hf_discovery import HuggingFaceDiscovery @@ -375,6 +375,7 @@ def choose_candidate_for_gpu_detailed( preflight_advisor: CandidatePreflightAdvisor | None = None, require_official_config: bool = False, config_optimizer: SafeConfigOptimizer | None = None, + excluded_frameworks: set[str] | None = None, ) -> tuple[CandidateModel | None, str | None]: last_rejection_reason: str | None = None for task_type in task_types: @@ -397,6 +398,14 @@ def choose_candidate_for_gpu_detailed( incumbent_frameworks=compatible_frameworks, inspection=inspection, ) + if excluded_frameworks: + available_frameworks = [ + framework for framework in compatible_frameworks + if framework not in excluded_frameworks + ] + if compatible_frameworks and not available_frameworks: + last_rejection_reason = "prior_failed_framework_no_alternative" + compatible_frameworks = available_frameworks if not compatible_frameworks: continue @@ -560,8 +569,10 @@ def process_model_for_candidates( if hasattr(modelhub_client, "model_submission_precheck"): precheck = modelhub_client.model_submission_precheck(model.repo_id) processed_gpus = set(precheck.get("processedGpus") or set()) + failed_frameworks_by_gpu = precheck.get("failedFrameworksByGpu") or {} else: processed_gpus = modelhub_client.processed_gpus_for_model(model.repo_id) + failed_frameworks_by_gpu = {} except Exception as exc: return ( [], @@ -580,8 +591,8 @@ def process_model_for_candidates( preferred_target_gpus = target_gpus[:MAX_ROUTES_PER_MODEL] fallback_target_gpus = target_gpus[MAX_ROUTES_PER_MODEL:] - def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str]]]: - pending: list[tuple[str, list[str]]] = [] + def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str], set[str]]]: + pending: list[tuple[str, list[str], set[str]]] = [] for target_gpu in route_target_gpus: if submission_exclusion_store is not None and submission_exclusion_store.is_blocked(model.repo_id, target_gpu): skipped.append( @@ -591,9 +602,45 @@ def process_model_for_candidates( if target_gpu in processed_gpus: skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"}) continue + if target_gpu in failed_frameworks_by_gpu and market_intelligence is None: + skipped.append({ + "repoId": model.repo_id, + "targetGpu": target_gpu, + "reason": "prior_failed_gpu_requires_public_framework_evidence", + }) + continue + failed_frameworks = set(failed_frameworks_by_gpu.get(target_gpu) or set()) + if ( + failed_frameworks + and market_intelligence is not None + and hasattr(market_intelligence, "has_vetted_alternative_framework") + and allowed_task_types_set + and all(market_intelligence.has_framework_route(task_type, target_gpu) for task_type in allowed_task_types_set) + and not any( + market_intelligence.has_vetted_alternative_framework( + task_type, target_gpu, failed_frameworks + ) + for task_type in allowed_task_types_set + ) + ): + skipped.append({ + "repoId": model.repo_id, + "targetGpu": target_gpu, + "reason": "prior_failed_framework_no_alternative", + }) + continue if outcome_tracker and outcome_tracker.is_model_gpu_failed(model.repo_id, target_gpu): skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"}) continue + if preflight_advisor is not None and model.file_size: + memory_gib = preflight_advisor.gpu_memory_gib.get(target_gpu) + if memory_gib and model.file_size / (1024**3) * MODEL_LOAD_OVERHEAD > memory_gib: + skipped.append({ + "repoId": model.repo_id, + "targetGpu": target_gpu, + "reason": "preflight_predicted_oom", + }) + continue exact_task_types = list(allowed_task_types_set) if official_registry is not None: @@ -632,7 +679,11 @@ def process_model_for_candidates( } ) continue - pending.append((target_gpu, compatible_task_types)) + pending.append(( + target_gpu, + compatible_task_types, + set(failed_frameworks_by_gpu.get(target_gpu) or set()), + )) return pending pending_task_types_by_gpu = prepare_pending_routes(preferred_target_gpus) @@ -649,9 +700,9 @@ def process_model_for_candidates( except Exception as exc: return [], skipped, [{"repoId": model.repo_id, "reason": str(exc)}] - def evaluate_pending_routes(pending_routes: list[tuple[str, list[str]]]) -> int: + def evaluate_pending_routes(pending_routes: list[tuple[str, list[str], set[str]]]) -> int: before_count = len(candidates) - for target_gpu, task_types in pending_routes: + for target_gpu, task_types, excluded_frameworks in pending_routes: if len(candidates) - before_count >= MAX_ROUTES_PER_MODEL: break best, preflight_reason = choose_candidate_for_gpu_detailed( @@ -664,6 +715,7 @@ def process_model_for_candidates( preflight_advisor=preflight_advisor, require_official_config=official_registry is not None, config_optimizer=config_optimizer, + excluded_frameworks=excluded_frameworks, ) if best is None: reason = ( @@ -677,6 +729,8 @@ def process_model_for_candidates( skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason}) continue record = candidate_to_record(best) + if excluded_frameworks: + record["previousFailedFrameworks"] = sorted(excluded_frameworks) if outcome_tracker is not None: record["transformersPrerequisiteSatisfied"] = ( outcome_tracker.has_non_transformers_failure(model.repo_id) @@ -753,12 +807,29 @@ def build_adaptive_scan_stages( "limit": max_models, } ) + if allow_older_than_recent_window: + # ModelScope only exposes the first 3,000 raw rows for a sort order. + # Downloads ordering reaches a different older, established slice + # after the recent-update slice has produced too few candidates. + stages.append( + { + "name": "popular_history", + "updatedAfter": None, + "limit": min(max_models, 200), + "sort": "downloads", + "rotatePages": True, + } + ) deduped: list[dict[str, Any]] = [] - seen: set[tuple[str | None, int]] = set() + seen: set[tuple[str | None, int, str]] = set() for stage in stages: updated_after = stage["updatedAfter"] - signature = (updated_after.isoformat() if updated_after is not None else None, int(stage["limit"])) + signature = ( + updated_after.isoformat() if updated_after is not None else None, + int(stage["limit"]), + str(stage.get("sort") or "last_modified"), + ) if signature in seen: continue seen.add(signature) @@ -877,6 +948,21 @@ def submit_candidate( "candidate": candidate, "reason": "already_processed_for_gpu", } + failed_frameworks = set( + (precheck.get("failedFrameworksByGpu") or {}).get(candidate["targetGpu"]) or set() + ) + if candidate["framework"] in failed_frameworks: + return { + "outcome": "duplicate", + "candidate": candidate, + "reason": "prior_failed_framework_for_gpu", + } + if failed_frameworks and not candidate.get("frameworkEvidenceQualified"): + return { + "outcome": "framework_evidence_deferred", + "candidate": candidate, + "reason": "prior_failed_gpu_requires_public_framework_evidence", + } payload = { "modelAddress": candidate["modelAddress"], @@ -1442,6 +1528,10 @@ def run_submission( "min_downloads": args.min_downloads, "updated_after": stage_updated_after, } + if stage.get("sort"): + query_kwargs["sort"] = stage["sort"] + if stage.get("rotatePages"): + query_kwargs["rotate_pages"] = True try: models = hf_discovery.list_recent_models( **query_kwargs, @@ -1501,6 +1591,7 @@ def run_submission( scanned_model_count += processed_count stage_summary = { "name": stage["name"], + "sort": stage.get("sort") or "last_modified", "updatedAfter": stage_updated_after.isoformat() if stage_updated_after is not None else None, "requestedLimit": stage_limit, "discoveredModels": len(models), @@ -1648,10 +1739,20 @@ def run_submission( { "repoId": candidate["repoId"], "targetGpu": candidate["targetGpu"], - "reason": "already_validating_on_platform", + "reason": result.get("reason") or "already_validating_on_platform", } ) continue + if result["outcome"] == "framework_evidence_deferred": + batch_policy_skipped_candidates.append(candidate) + skipped.append({ + "repoId": candidate["repoId"], + "targetGpu": candidate["targetGpu"], + "framework": candidate["framework"], + "reason": result["reason"], + "deferred": True, + }) + continue if result["outcome"] == "age_policy_deferred": batch_policy_skipped_candidates.append(candidate) skipped.append( diff --git a/modelhub_submmit_api/market_intelligence.py b/modelhub_submmit_api/market_intelligence.py index ede1cbca15..d6f6bfb953 100644 --- a/modelhub_submmit_api/market_intelligence.py +++ b/modelhub_submmit_api/market_intelligence.py @@ -859,6 +859,20 @@ class MarketIntelligenceManager: ) return any(bool(item.get("officialConfigValid", False)) for item in rows.values()) + def has_vetted_alternative_framework( + self, task_type: str, target_gpu: str, excluded_frameworks: set[str] + ) -> bool: + rows = ( + (((self.state or {}).get("frameworkStats") or {}).get(task_type) or {}).get(target_gpu) + or {} + ) + return any( + framework not in excluded_frameworks + and bool(item.get("officialConfigValid", False)) + and self._framework_evidence(task_type, target_gpu, framework)["qualified"] + for framework, item in rows.items() + ) + def ensure_framework_route(self, client: Any, task_type: str, target_gpu: str) -> None: if self.has_framework_route(task_type, target_gpu): return diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 127c05cd7e..2e43a76ed6 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -36,9 +36,41 @@ def parse_model_submission_precheck(payload: Any) -> dict[str, Any]: if not isinstance(verify_result, dict): raise ModelHubAPIError("Community model precheck verifyResult is invalid", payload=payload) + blocked_gpus: set[str] = set() + failed_frameworks_by_gpu: dict[str, set[str]] = {} + for gpu, gpu_result in verify_result.items(): + name = str(gpu) + if not isinstance(gpu_result, dict): + blocked_gpus.add(name) + continue + status = str(gpu_result.get("result") or "").strip() + records = gpu_result.get("records") + if status == "未验证" and isinstance(records, list) and not records: + continue + if status == "验证失败" and isinstance(records, list) and records: + frameworks: set[str] = set() + for record in records: + if not isinstance(record, dict): + break + framework = str(record.get("framework") or "").strip() + try: + failed = int(record.get("verifyResult")) < 0 + except (TypeError, ValueError): + failed = False + if not framework or not failed: + break + frameworks.add(framework) + else: + failed_frameworks_by_gpu[name] = frameworks + continue + # Verified, validating and unknown states remain closed. A failed GPU + # opens only for a different framework with explicit failed records. + blocked_gpus.add(name) + return { "isInDB": data.get("isInDB") is True, - "processedGpus": set(str(gpu) for gpu in verify_result), + "processedGpus": blocked_gpus, + "failedFrameworksByGpu": failed_frameworks_by_gpu, "verifyResult": verify_result, "modelInfo": data.get("modelInfo") if isinstance(data.get("modelInfo"), dict) else {}, } @@ -73,10 +105,7 @@ class ModelHubClient: return parse_model_submission_precheck(self.search_by_model_id(model_id, force_refresh=force_refresh)) def is_model_processed_for_gpu(self, model_id: str, target_gpu: str) -> bool: - gpu_result = self.get_verify_result_map(model_id).get(target_gpu) - if gpu_result is None: - return False - return "result" in gpu_result or "records" in gpu_result + return target_gpu in self.model_submission_precheck(model_id)["processedGpus"] def get_verify_result_map(self, model_id: str) -> dict[str, Any]: payload = self.search_by_model_id(model_id) diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index a084ff997b..acff228d8d 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -9,7 +9,7 @@ from collections import deque from pathlib import Path from typing import Any, Callable -from common import read_json, read_jsonl, utc_now, write_json +from common import parse_datetime, read_json, read_jsonl, utc_now, write_json from daily_runner import DEFAULT_DAILY_RUNS_DIR, log, run_daily_batches from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH from hf_discovery import HuggingFaceDiscovery @@ -445,6 +445,7 @@ def _advance_architecture_history_backfill( batch_tasks: list[dict[str, Any]] = [] page_calls = 0 page_errors = 0 + last_error = "" for _ in range(pages_per_batch): incomplete = [ @@ -468,7 +469,9 @@ def _advance_architecture_history_backfill( current=current_page, page_size=page_size, only_mine=True, - end_time=str(progress.get("cutoffAt") or now.isoformat()), + # Pass a datetime so ModelHubClient formats the platform's + # YYYY-MM-DD HH:MM:SS query instead of an ISO timestamp. + end_time=parse_datetime(progress.get("cutoffAt")) or now, ) data = payload.get("data") or {} records = [item for item in (data.get("records") or []) if isinstance(item, dict)] @@ -500,6 +503,7 @@ def _advance_architecture_history_backfill( page_errors += 1 state["listingErrors"] = int(state.get("listingErrors") or 0) + 1 state["lastError"] = f"{type(exc).__name__}: {exc}"[:500] + last_error = state["lastError"].replace("\n", " ") break batch_summary: dict[str, Any] = { @@ -558,7 +562,8 @@ def _advance_architecture_history_backfill( f"failure_logs={int(batch_summary.get('enrichmentAttempts') or 0)} " f"accounts={completed_accounts}/{len(clients)} " f"unique_total={int(progress.get('uniqueRecords') or 0)} " - "retention=decision_state_only" + f"retention=decision_state_only" + f"{f' error={last_error}' if last_error else ''}" ) return progress diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 40335c5068..5f567de2a9 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.09.18.1" +AGENT_VERSION = "2026.09.20.1" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 98254ce361..2ccbf8b11d 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -544,7 +544,8 @@ class ClientPoolConcurrencyTests(unittest.TestCase): self.assertEqual(2, summary["submittedCount"]) self.assertEqual(len(supported_gpus), summary["skipReasonCounts"]["already_processed_for_gpu"]) - self.assertEqual("all_history", summary["scanStages"][-1]["name"]) + self.assertEqual("popular_history", summary["scanStages"][-1]["name"]) + self.assertEqual("downloads", summary["scanStages"][-1]["sort"]) self.assertIsNone(discovery.calls[-1]) def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None: diff --git a/tests/test_market_intelligence.py b/tests/test_market_intelligence.py index 45e8db9daf..9199ef48cd 100644 --- a/tests/test_market_intelligence.py +++ b/tests/test_market_intelligence.py @@ -115,6 +115,14 @@ class MarketIntelligenceTests(unittest.TestCase): self.assertTrue( state["frameworkStats"]["text-generation"]["fast"]["transformers"]["officialConfigValid"] ) + self.assertTrue( + manager.has_vetted_alternative_framework("text-generation", "fast", {"vllm"}) + ) + self.assertFalse( + manager.has_vetted_alternative_framework( + "text-generation", "fast", {"vllm", "transformers"} + ) + ) def test_snapshot_cache_prevents_repeated_public_api_scans(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: diff --git a/tests/test_modelscope_discovery.py b/tests/test_modelscope_discovery.py index e174a3527b..b3f8e5e4c0 100644 --- a/tests/test_modelscope_discovery.py +++ b/tests/test_modelscope_discovery.py @@ -93,6 +93,31 @@ class ModelScopeDiscoveryTests(unittest.TestCase): self.assertEqual([], models) self.assertEqual(50, http_client.queries[0]["page_size"]) + def test_popular_sort_uses_separate_page_cache(self) -> None: + http_client = RecordingHttpClient() + discovery = HuggingFaceDiscovery(http_client=http_client) # type: ignore[arg-type] + + discovery.list_recent_models(pipeline_tags=["text-generation"], limit=50, min_downloads=0) + discovery.list_recent_models( + pipeline_tags=["text-generation"], limit=50, min_downloads=0, sort="downloads" + ) + self.assertEqual(["last_modified", "downloads"], [item["sort"] for item in http_client.queries]) + + def test_popular_scan_rotates_bounded_page_windows(self) -> None: + http_client = RecordingHttpClient() + discovery = HuggingFaceDiscovery(http_client=http_client) # type: ignore[arg-type] + + for _ in range(2): + discovery.list_recent_models( + pipeline_tags=["text-generation"], + limit=200, + min_downloads=50, + sort="downloads", + rotate_pages=True, + ) + + self.assertEqual([1, 5], [item["page_number"] for item in http_client.queries]) + def test_rate_limited_page_keeps_previous_pages_and_resumes_from_cache(self) -> None: http_client = RateLimitedPageClient() discovery = HuggingFaceDiscovery( diff --git a/tests/test_poll_policy.py b/tests/test_poll_policy.py index 7645106239..7baeb30cc3 100644 --- a/tests/test_poll_policy.py +++ b/tests/test_poll_policy.py @@ -27,10 +27,12 @@ class PollPolicyTests(unittest.TestCase): def test_architecture_history_backfill_resumes_and_keeps_only_decision_state(self) -> None: class PagedHistoryClient: calls: list[int] = [] + end_times: list[datetime] = [] def list_tasks_page(self, **kwargs): # noqa: ANN003, ANN201 current = int(kwargs["current"]) self.calls.append(current) + self.end_times.append(kwargs["end_time"]) records = { 1: [ { @@ -101,6 +103,8 @@ class PollPolicyTests(unittest.TestCase): self.assertTrue(second["complete"]) self.assertNotIn("seenTaskIds", second) self.assertEqual([1, 2], client.calls) + self.assertTrue(all(isinstance(value, datetime) for value in client.end_times)) + self.assertEqual(client.end_times[0], client.end_times[1]) self.assertEqual(3, restored.get_stats_report()["terminalRecords"]) self.assertEqual([], json.loads(outcomes.read_text(encoding="utf-8") or "[]")) persisted_progress = progress_path.read_text(encoding="utf-8") diff --git a/tests/test_submission_safety.py b/tests/test_submission_safety.py index c3cfcc7cfd..3feb372997 100644 --- a/tests/test_submission_safety.py +++ b/tests/test_submission_safety.py @@ -13,6 +13,7 @@ if str(PACKAGE_DIR) in sys.path: sys.path.insert(0, str(PACKAGE_DIR)) from main import build_parser, one_candidate_per_model, process_model_for_candidates, run_submission, submit_candidate # noqa: E402 +from candidate_preflight import CandidatePreflightAdvisor # noqa: E402 from modelhub_client import ( # noqa: E402 ModelHubAPIError, ModelHubClientPool, @@ -68,6 +69,23 @@ class ExactGpuProcessedClient: return {"code": 0, "data": {"id": "should-not-submit"}} +class FailedFrameworkClient: + def __init__(self) -> None: + self.add_calls = 0 + + @staticmethod + def model_submission_precheck(_model_id: str, *, force_refresh: bool = False) -> dict: + del force_refresh + return { + "processedGpus": set(), + "failedFrameworksByGpu": {"Cambricon_mlu-370-x4": {"vllm"}}, + } + + def add_task(self, _payload: dict) -> dict: + self.add_calls += 1 + return {"code": 0, "data": {"id": str(self.add_calls)}} + + class AllGpuOfficialRegistry: @staticmethod def task_types_for(_client, *, model_address: str, model_last_modified: str | None, gpu: str) -> list[str]: @@ -100,7 +118,11 @@ class AllGpuVettedMarket: @staticmethod def framework_metadata(_task_type: str, _target_gpu: str, _framework: str) -> dict: - return {"frameworkCombinedScore": 1.0, "frameworkOfficialConfigValid": True} + return { + "frameworkCombinedScore": 1.0, + "frameworkOfficialConfigValid": True, + "frameworkEvidenceQualified": True, + } @staticmethod def gpu_metadata(_target_gpu: str) -> dict: @@ -113,6 +135,15 @@ class SeventhGpuFallbackMarket(AllGpuVettedMarket): return target_gpu == "gpu-7" +class TwoFrameworkMarket(AllGpuVettedMarket): + @staticmethod + def selectable_frameworks( + *, task_type: str, target_gpu: str, incumbent_frameworks: list[str], inspection + ) -> list[str]: + del task_type, target_gpu, incumbent_frameworks, inspection + return ["vllm", "vllm-mlu"] + + class UniquenessRejectingClient: @staticmethod def add_task(_payload: dict) -> dict: @@ -230,6 +261,155 @@ class SubmissionSafetyTests(unittest.TestCase): self.assertEqual("already_processed_for_gpu", skipped[0]["reason"]) self.assertEqual([], failed) + def test_precheck_distinguishes_unverified_failed_and_active_gpus(self) -> None: + payload = { + "data": { + "isInDB": True, + "verifyResult": { + "free": {"result": "未验证", "records": []}, + "failed": { + "result": "验证失败", + "records": [{"framework": "vllm", "verifyResult": -1}], + }, + "running": {"result": "验证中", "records": []}, + "success": { + "result": "已验证", + "records": [{"framework": "vllm", "verifyResult": 1}], + }, + "unknown": {"result": "未知状态", "records": []}, + "failed_without_framework": { + "result": "验证失败", + "records": [{"verifyResult": -1}], + }, + }, + }, + } + precheck = parse_model_submission_precheck(payload) + self.assertEqual( + {"running", "success", "unknown", "failed_without_framework"}, + precheck["processedGpus"], + ) + self.assertEqual({"failed": {"vllm"}}, precheck["failedFrameworksByGpu"]) + + def test_unverified_gpu_becomes_candidate_but_validating_gpu_does_not(self) -> None: + class StatusClient: + def __init__(self, status: str) -> None: + self.status = status + + def model_submission_precheck(self, _model_id: str) -> dict: + return parse_model_submission_precheck({ + "data": { + "verifyResult": { + "Vastai_va16": {"result": self.status, "records": []} + } + } + }) + + def collect(status: str): + return process_model_for_candidates( + model=SafeDiscovery().model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=StatusClient(status), # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=["Vastai_va16"], + allowed_task_types=["text-generation"], + ) + + open_candidates, _, _ = collect("未验证") + active_candidates, active_skips, _ = collect("验证中") + self.assertEqual(["Vastai_va16"], [item["targetGpu"] for item in open_candidates]) + self.assertEqual([], active_candidates) + self.assertEqual("already_processed_for_gpu", active_skips[0]["reason"]) + + def test_failed_gpu_requires_an_alternative_framework(self) -> None: + model = SafeDiscovery().model + client = FailedFrameworkClient() + candidates, skipped, failed = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=["Cambricon_mlu-370-x4"], + allowed_task_types=["text-generation"], + market_intelligence=TwoFrameworkMarket(), # type: ignore[arg-type] + official_registry=AllGpuOfficialRegistry(), # type: ignore[arg-type] + ) + self.assertEqual([], skipped) + self.assertEqual([], failed) + self.assertEqual(["vllm-mlu"], [item["framework"] for item in candidates]) + self.assertEqual(["vllm"], candidates[0]["previousFailedFrameworks"]) + + blocked = submit_candidate({**candidates[0], "framework": "vllm"}, client) # type: ignore[arg-type] + accepted = submit_candidate(candidates[0], client) # type: ignore[arg-type] + self.assertEqual("duplicate", blocked["outcome"]) + self.assertEqual("prior_failed_framework_for_gpu", blocked["reason"]) + self.assertEqual("submitted", accepted["outcome"]) + self.assertEqual(1, client.add_calls) + + unvetted = submit_candidate( + {**candidates[0], "frameworkEvidenceQualified": False}, client # type: ignore[arg-type] + ) + self.assertEqual("framework_evidence_deferred", unvetted["outcome"]) + self.assertEqual(1, client.add_calls) + + no_routes, no_route_skips, _ = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=["Cambricon_mlu-370-x4"], + allowed_task_types=["text-generation"], + market_intelligence=AllGpuVettedMarket(), # type: ignore[arg-type] + official_registry=AllGpuOfficialRegistry(), # type: ignore[arg-type] + ) + self.assertEqual([], no_routes) + self.assertEqual("prior_failed_framework_no_alternative", no_route_skips[-1]["reason"]) + + no_market_routes, no_market_skips, _ = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=["Cambricon_mlu-370-x4"], + allowed_task_types=["text-generation"], + ) + self.assertEqual([], no_market_routes) + self.assertEqual( + "prior_failed_gpu_requires_public_framework_evidence", + no_market_skips[-1]["reason"], + ) + + def test_published_size_avoids_official_calls_for_certain_oom(self) -> None: + class CountingRegistry(AllGpuOfficialRegistry): + calls = 0 + + def task_types_for(self, *_args, **_kwargs) -> list[str]: + self.calls += 1 + return ["text-generation"] + + registry = CountingRegistry() + model = HFModelSummary( + repo_id="owner/oversized", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + file_size=100 * 1024**3, + ) + candidates, skipped, failed = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=OtherGpuProcessedClient(), # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=["MetaX_c-500"], + allowed_task_types=["text-generation"], + preflight_advisor=CandidatePreflightAdvisor(), + official_registry=registry, # type: ignore[arg-type] + ) + self.assertEqual([], candidates) + self.assertEqual("preflight_predicted_oom", skipped[0]["reason"]) + self.assertEqual([], failed) + self.assertEqual(0, registry.calls) + def test_lookup_failure_is_fail_closed(self) -> None: pool = ModelHubClientPool([FailingLookupClient()]) # type: ignore[arg-type] with self.assertRaises(ModelHubAPIError):