feat: calibrate account capacity weekly

README.md

@@ -32,7 +32,7 @@ Optional tuning:
 - The hosted entrypoint always uses `--max-submits-per-run 0` so stale deployment
   settings cannot restrict a refill cycle to five submissions.
 - `MODELHUB_AGENT_ACTIVE_TASK_CAP` default `100` per account
-- `MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES` default `3`
+- `MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES` is a legacy setting ignored by the hosted worker; it uses weekly calibration only
 - `MODELHUB_CAPACITY_STATE_PATH` default `.modelhub_state/account_capacity.json`
 - `MODELHUB_AGENT_ACTIVE_COUNTS_TTL_SECONDS` default `15`
 - `MODELHUB_AGENT_RESERVATION_TTL_SECONDS` default `120`
@@ -228,13 +228,25 @@ or 500, older models stop at positions 95, 195, or 495 respectively. Old-model
 reservations are made under the same account lock as capacity reservations, so
 concurrent submissions cannot enter the reserved suffix. Once every account's
 old-model allowance is exhausted, discovery is capped at the seven-day window.
-Unknown model timestamps are treated as old for new submissions. When capacity
+Unknown model timestamps are treated as old for new submissions. When weekly
-probing raises an account's known limit, its old-model boundary moves with it.
+calibration raises an account's known limit, its old-model boundary moves with it.
 If an account's active-count read fails, old-model admission is fail-closed for
 that account while other accounts remain eligible. Date-deferred candidates are
 recorded as `age_policy_skipped`, release their temporary claim, and do not enter
 failure statistics or permanent exclusions.
 
+Capacity calibration runs only on Monday from 10:00 to 10:59 Asia/Shanghai.
+During this window the worker targets each account in turn and temporarily
+borrows its five reserved positions, using only normally vetted candidates.
+An explicit capacity-full response ends that account's calibration; the
+measured limit and completed week are persisted in `account_capacity.json`.
+Rate limiting is not treated as proof of account capacity. If no safe candidates
+are found, calibration waits ten minutes before retrying within the same window.
+If the window ends or the service was offline throughout it, it does not
+continue or catch up outside the window. Normal capacity-minus-five
+admission resumes without cancelling already accepted tasks; reserved positions
+reappear as tasks finish.
+
 Model age is admission-only. Startup and scheduled queue cleanup never cancel a
 task because it is older than seven days, beyond a capacity-minus-five boundary,
 or above a temporarily reduced platform limit. Existing waiting and running
@@ -451,11 +463,29 @@ 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.
+
+Version `2026.09.20.2` replaces routine capacity probes with per-account
+calibration each Monday 10:00-10:59 Asia/Shanghai. During that window only,
+the worker may borrow the five recent-model positions to measure a new limit;
+ordinary capacity-minus-five admission resumes afterward.
+
 ## 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 tag -a agent-v36 -m "ModelHub agent 2026.09.20.2"
-git push origin main agent-v34
+git push origin main agent-v36
 ```

modelhub_submmit_api/README.md

@@ -115,22 +115,30 @@ 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.
 - ModelScope list pages are paced and cached for 15 minutes. HTTP 429 responses use exponential
   backoff and `Retry-After`; pages already downloaded remain usable and the failed page is retried
   on the next cycle.
-- Every third poll cycle, a full account gets one controlled capacity probe. A successful probe
+- Each Monday 10:00-10:59 Asia/Shanghai, the worker calibrates accounts one at a time.
-  raises that account's persisted known limit; a capacity rejection enters cooldown.
+  It temporarily borrows the final five reserved positions and probes beyond the known
+  limit until explicit account-capacity rejection, then persists that account's limit and
+  resumes the normal capacity-minus-five policy. A restart in the same window does not
+  repeat accounts already calibrated; a missed window is not made up later.
 - Each `[scan]` log records the discovery stage and candidate yield. The final `[daily] wave_done`
   log includes `skip_reasons`, making empty candidate pools distinguishable from API failures.
 - At startup and every 120 poll cycles, active tasks are checked with the same recursive-size
@@ -139,7 +147,7 @@ bash run_poll.sh --dry-run
   Incomplete size/capacity evidence is never used for cancellation.
 - Models older than seven days may occupy only the current account capacity minus its final 5
   positions. The account pool enforces this per-account admission boundary atomically and updates
-  it when capacity probing discovers a higher limit. A failed active-count read makes that account
+  it when weekly calibration discovers a higher limit. A failed active-count read makes that account
   ineligible for older models while other readable accounts are still tried. Recent models may use
   all available positions. Age policy never cancels an existing waiting or running task; startup
   and scheduled cleanup remain limited to deterministic OOM and learned architecture evidence.

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):
+        page_count = (max_items + page_size - 1) // page_size
-            cache_key = ("*" if filter_disabled else task_tag, page_number, 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} "

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
@@ -258,7 +258,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--capacity-probe-interval-cycles",
         type=int,
-        default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "3")),
+        default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "0")),
         help=argparse.SUPPRESS,
     )
     parser.add_argument("--hf-base-url", default=os.getenv("MODELSCOPE_BASE_URL", "https://modelscope.cn"), help=argparse.SUPPRESS)
@@ -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
 
@@ -508,6 +517,9 @@ def resolve_submit_concurrency(
     modelhub_client: ModelHubClient | ModelHubClientPool | Any,
     planned_submit_count: int,
 ) -> int:
+    if hasattr(modelhub_client, "weekly_calibration_active") and modelhub_client.weekly_calibration_active():
+        # Pinning and capacity evidence require one in-flight task per account.
+        return 1
     explicit_workers = int(getattr(args, "submit_concurrency", 0) or 0)
     if explicit_workers > 0:
         return explicit_workers
@@ -560,8 +572,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 +594,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]]]:
+    def prepare_pending_routes(route_target_gpus: list[str]) -> list[tuple[str, list[str], set[str]]]:
-        pending: list[tuple[str, list[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 +605,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 +682,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 +703,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 +718,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 +732,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 +810,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 +951,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"],
@@ -964,6 +1053,17 @@ def submit_candidate(
                 "candidate": candidate,
                 "reason": "account_capacity_saturated",
             }
+        if exc.code == 429:
+            print(
+                f"[submit] deferred repo={candidate['repoId']} gpu={candidate['targetGpu']} "
+                f"framework={candidate['framework']} reason=platform_rate_limited",
+                flush=True,
+            )
+            return {
+                "outcome": "rate_limit_deferred",
+                "candidate": candidate,
+                "reason": "platform_rate_limited",
+            }
         print(
             f"[submit] failed repo={candidate['repoId']} gpu={candidate['targetGpu']} "
             f"framework={candidate['framework']} reason={exc}",
@@ -1064,11 +1164,12 @@ def run_submission(
         clients = [ModelHubClient(token=token, base_url=args.modelhub_base_url) for token in token_values]
         modelhub_client = ModelHubClientPool(
             clients,
-            capacity_probe_interval_cycles=max(0, int(getattr(args, "capacity_probe_interval_cycles", 3) or 0)),
+            capacity_probe_interval_cycles=0,
             capacity_state_path=Path(getattr(args, "capacity_state_path", DEFAULT_CAPACITY_STATE_PATH)),
             recent_model_reserve_slots=max(0, int(getattr(args, "recent_model_reserve_slots", 5) or 0)),
             recent_model_days=max(1, int(getattr(args, "recent_model_days", 7) or 7)),
         )
+        modelhub_client.configure_weekly_calibration(now)
 
     official_registry: OfficialCapabilityRegistry | None = None
     official_summary: dict[str, Any] = {"enabled": False}
@@ -1413,10 +1514,16 @@ def run_submission(
     old_model_queue_thresholds: list[int | None] | None = None
     if hasattr(modelhub_client, "old_model_queue_thresholds"):
         old_model_queue_thresholds = list(modelhub_client.old_model_queue_thresholds())
+    weekly_calibration = (
+        modelhub_client.weekly_calibration_status()
+        if hasattr(modelhub_client, "weekly_calibration_status") else {}
+    )
     allow_older_models_for_scan = old_model_slots_before_scan is None or old_model_slots_before_scan > 0
     print(
-        f"[age-policy] mode=admission_only recent_days={recent_model_days} "
+        f"[age-policy] mode={'weekly_capacity_calibration' if weekly_calibration.get('targetAccount') else 'admission_only'} "
+        f"recent_days={recent_model_days} "
         f"reserve_recent_slots={recent_model_reserve_slots} "
+        f"calibration_account={weekly_calibration.get('targetAccount') or 'n/a'} "
         f"account_thresholds={','.join('n/a' if value is None else str(value) for value in old_model_queue_thresholds) if old_model_queue_thresholds is not None else 'n/a'} "
         f"old_model_slots={old_model_slots_before_scan if old_model_slots_before_scan is not None else 'n/a'} "
         f"scan_older={'on' if allow_older_models_for_scan else 'off'}",
@@ -1442,6 +1549,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 +1612,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 +1760,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"] in {"framework_evidence_deferred", "rate_limit_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(
@@ -1805,13 +1927,15 @@ def run_submission(
         "marketIntelligence": market_summary,
         "candidatePreflight": preflight_summary,
         "agePolicy": {
-            "mode": "admission_only",
+            "mode": "weekly_capacity_calibration" if weekly_calibration.get("targetAccount") else "admission_only",
             "recentModelDays": recent_model_days,
             "recentModelReserveSlots": recent_model_reserve_slots,
             "oldModelSubmitThresholds": old_model_queue_thresholds,
             "oldModelSlotsBeforeScan": old_model_slots_before_scan,
             "olderHistoryScanEnabled": allow_older_models_for_scan,
+            "calibrationAccount": weekly_calibration.get("targetAccount"),
         },
+        "weeklyCalibration": weekly_calibration,
         "scanLimit": scan_limit,
         "rawCandidateGoal": raw_candidate_goal,
         "candidatePoolLimit": candidate_pool_limit,

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

modelhub_submmit_api/modelhub_client.py

@@ -8,6 +8,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from common import format_modelhub_datetime, parse_datetime, read_json, runtime_instance_id, utc_now, write_json
 from defaults import EMBEDDED_MODELHUB_XC_TOKEN
@@ -36,9 +37,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 +106,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)
+        return target_gpu in self.model_submission_precheck(model_id)["processedGpus"]
-        if gpu_result is None:
-            return False
-        return "result" in gpu_result or "records" in gpu_result
 
     def get_verify_result_map(self, model_id: str) -> dict[str, Any]:
         payload = self.search_by_model_id(model_id)
@@ -281,6 +311,17 @@ class ModelHubClient:
 
         return active_count
 
+    def count_active_tasks_exact(self) -> int:
+        """Use the platform's filtered totals instead of paging all history."""
+        total = 0
+        for status in ("waiting", "running"):
+            page = self.list_tasks_page(current=1, page_size=1, only_mine=True, status=status)
+            data = page.get("data")
+            if not isinstance(data, dict) or "total" not in data:
+                raise ModelHubAPIError(f"Active-task total is unavailable for status={status}")
+            total += max(0, int(data["total"]))
+        return total
+
     def _request(
         self,
         method: str,
@@ -380,8 +421,6 @@ DEFAULT_CAPACITY_STATE_PATH = Path(".modelhub_state/account_capacity.json")
 
 
 def is_capacity_error(error: ModelHubAPIError) -> bool:
-    if error.code in {409, 429}:
-        return True
     message = str(error).strip().lower()
     if isinstance(error.payload, dict):
         message = f"{message} {error.payload.get('message') or ''}".lower()
@@ -418,7 +457,7 @@ class ModelHubClientPool:
         active_counts_ttl: float | None = None,
         reservation_ttl: float | None = None,
         instance_id: str | None = None,
-        capacity_probe_interval_cycles: int = 3,
+        capacity_probe_interval_cycles: int = 0,
         capacity_probe_cooldown_cycles: int = 3,
         capacity_state_path: Path | str | None = None,
         recent_model_reserve_slots: int | None = None,
@@ -443,7 +482,13 @@ class ModelHubClientPool:
         self.recent_model_days = max(1, int(configured_recent_days))
         self._capacity_state_path = Path(capacity_state_path) if capacity_state_path else None
         self._account_keys = [self._account_key(client, index) for index, client in enumerate(clients)]
+        self._weekly_calibration_weeks: list[str | None] = [None for _ in clients]
+        self._weekly_calibration_results: list[str | None] = [None for _ in clients]
         self._account_caps = self._load_account_caps(self.active_task_cap)
+        self._weekly_calibration_week: str | None = None
+        self._weekly_calibration_window = False
+        self._weekly_calibration_deadline = 0.0
+        self._weekly_no_candidate_retry_at = 0.0
         self._capacity_probe_interval_cycles = max(0, int(capacity_probe_interval_cycles))
         self._capacity_probe_cooldown_cycles = max(1, int(capacity_probe_cooldown_cycles))
         self._capacity_probe_cycle = 0
@@ -492,10 +537,15 @@ class ModelHubClientPool:
                     stored = payload.get("accounts") or {}
             except (FileNotFoundError, ValueError):
                 pass
-        return [
+        caps: list[int] = []
-            max(1, int((stored.get(key) or {}).get("knownCap") or default_cap))
+        for index, key in enumerate(self._account_keys):
-            for key in self._account_keys
+            account = stored.get(key) or {}
-        ]
+            if not isinstance(account, dict):
+                account = {}
+            caps.append(max(1, int(account.get("knownCap") or default_cap)))
+            self._weekly_calibration_weeks[index] = account.get("calibrationWeek") or None
+            self._weekly_calibration_results[index] = account.get("calibrationResult") or None
+        return caps
 
     def _persist_account_caps(self) -> None:
         if self._capacity_state_path is None:
@@ -505,7 +555,12 @@ class ModelHubClientPool:
                 "version": 1,
                 "updatedAt": datetime.now().astimezone().isoformat(),
                 "accounts": {
-                    key: {"accountIndex": index + 1, "knownCap": self._account_caps[index]}
+                    key: {
+                        "accountIndex": index + 1,
+                        "knownCap": self._account_caps[index],
+                        "calibrationWeek": self._weekly_calibration_weeks[index],
+                        "calibrationResult": self._weekly_calibration_results[index],
+                    }
                     for index, key in enumerate(self._account_keys)
                 },
             }
@@ -523,6 +578,92 @@ class ModelHubClientPool:
                 and cycle_number % self._capacity_probe_interval_cycles == 0
             )
 
+    def configure_weekly_calibration(self, now: datetime) -> dict[str, Any]:
+        """Only Monday 10:00-10:59 in Beijing may borrow recent-model slots."""
+        local_now = now.astimezone(ZoneInfo("Asia/Shanghai"))
+        week = f"{local_now.isocalendar().year}-W{local_now.isocalendar().week:02d}"
+        window = local_now.weekday() == 0 and local_now.hour == 10
+        with self._state_lock:
+            if week != self._weekly_calibration_week:
+                self._weekly_no_candidate_retry_at = 0.0
+            self._weekly_calibration_week = week if window else None
+            self._weekly_calibration_window = window
+            self._weekly_calibration_deadline = (
+                time.monotonic()
+                + (local_now.replace(hour=11, minute=0, second=0, microsecond=0) - local_now).total_seconds()
+                if window else 0.0
+            )
+            target = self._weekly_target_locked()
+            return {
+                "window": window,
+                "week": week,
+                "targetAccount": target + 1 if target is not None else None,
+                "completedAccounts": sum(value == week for value in self._weekly_calibration_weeks),
+            }
+
+    def _weekly_target_locked(self) -> int | None:
+        if (
+            not self._weekly_calibration_window
+            or not self._weekly_calibration_week
+            or time.monotonic() >= self._weekly_calibration_deadline
+            or time.monotonic() < self._weekly_no_candidate_retry_at
+        ):
+            return None
+        for index in range(len(self.clients)):
+            if (
+                self._weekly_calibration_weeks[index] != self._weekly_calibration_week
+                and self._count_known[index]
+            ):
+                return index
+        return None
+
+    def weekly_calibration_active(self) -> bool:
+        self._refresh_active_counts()
+        with self._state_lock:
+            return self._weekly_target_locked() is not None
+
+    def weekly_calibration_status(self) -> dict[str, Any]:
+        self._refresh_active_counts()
+        with self._state_lock:
+            target = self._weekly_target_locked()
+            return {
+                "window": self._weekly_calibration_window and time.monotonic() < self._weekly_calibration_deadline,
+                "week": self._weekly_calibration_week,
+                "targetAccount": target + 1 if target is not None else None,
+                "completedAccounts": sum(
+                    value == self._weekly_calibration_week
+                    for value in self._weekly_calibration_weeks
+                ) if self._weekly_calibration_week else 0,
+            }
+
+    def finish_weekly_calibration(self, index: int, *, result: str) -> None:
+        with self._state_lock:
+            if not self._weekly_calibration_window or not self._weekly_calibration_week:
+                return
+            self._weekly_calibration_weeks[index] = self._weekly_calibration_week
+            self._weekly_calibration_results[index] = result
+            known_cap = self._account_caps[index]
+            active = self._effective_count_locked(index)
+            threshold = self._old_model_count_limit_locked(index)
+        self._persist_account_caps()
+        print(
+            f"[capacity-calibration] account={index + 1:02d} result={result} "
+            f"known_cap={known_cap} active={active} old_threshold={threshold}",
+            flush=True,
+        )
+
+    def pause_weekly_calibration_for_no_candidates(self) -> None:
+        with self._state_lock:
+            target = self._weekly_target_locked()
+            if target is None:
+                return
+            self._weekly_no_candidate_retry_at = time.monotonic() + 600
+        print(
+            f"[capacity-calibration] account={target + 1:02d} "
+            "status=paused reason=no_candidates retry_seconds=600",
+            flush=True,
+        )
+
     def account_capacity_limits(self) -> list[int]:
         with self._state_lock:
             return list(self._account_caps)
@@ -547,8 +688,10 @@ class ModelHubClientPool:
         with self._state_lock:
             return self._capacity_probe_enabled
 
-    def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int) -> int | None:
+    def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int | None) -> int | None:
         try:
+            if max_count is None and hasattr(client, "count_active_tasks_exact"):
+                return client.count_active_tasks_exact()
             return client.count_active_tasks(max_count=max_count, page_size=200)
         except Exception:
             return None
@@ -577,7 +720,25 @@ class ModelHubClientPool:
                     return
 
             with self._state_lock:
-                count_limits = [cap + 1 for cap in self._account_caps]
+                count_limits: list[int | None] = [cap + 1 for cap in self._account_caps]
+                weekly_target = self._weekly_target_locked()
+                if (
+                    weekly_target is None
+                    and self._weekly_calibration_window
+                    and time.monotonic() < self._weekly_calibration_deadline
+                    and time.monotonic() >= self._weekly_no_candidate_retry_at
+                ):
+                    weekly_target = next(
+                        (
+                            index for index, week in enumerate(self._weekly_calibration_weeks)
+                            if week != self._weekly_calibration_week
+                        ),
+                        None,
+                    )
+                if weekly_target is not None:
+                    # A cap+1 early stop only proves a lower bound. Weekly
+                    # calibration needs an exact active count for its target.
+                    count_limits[weekly_target] = None
 
             def _to_indexed_result(index: int, client: ModelHubClient) -> tuple[int, int | None]:
                 return index, self._safe_count_active_tasks(client, count_limits[index])
@@ -665,6 +826,10 @@ class ModelHubClientPool:
     def available_submit_slots(self) -> int:
         self._refresh_active_counts()
         with self._state_lock:
+            if self._weekly_target_locked() is not None:
+                # The weekly run may need to cross the previously known cap.
+                # Bound each scan; it can resume in the next cycle.
+                return 100
             normal_slots = sum(
                 max(0, self._account_caps[index] - self._effective_count_locked(index))
                 for index in range(len(self.clients))
@@ -752,10 +917,12 @@ class ModelHubClientPool:
         reserve_capacity_slots: int | None = None,
     ) -> tuple[int, int] | None:
         with self._state_lock:
+            weekly_target = self._weekly_target_locked()
             remaining_by_index = {
                 index: self._account_caps[index] - self._effective_count_locked(index)
                 for index in range(len(self.clients))
                 if index not in excluded
+                and (weekly_target is None or index == weekly_target)
                 and (
                     reserve_capacity_slots is None
                     or (
@@ -790,14 +957,19 @@ class ModelHubClientPool:
         reserve_capacity_slots: int | None = None,
     ) -> tuple[int, int] | None:
         with self._state_lock:
-            if not self._capacity_probe_enabled:
+            weekly_target = self._weekly_target_locked()
+            if weekly_target is None and not self._capacity_probe_enabled:
                 return None
             eligible = [
                 index
                 for index in range(len(self.clients))
                 if index not in excluded
-                and index not in self._capacity_probe_attempted
+                and (weekly_target is None or index == weekly_target)
-                and self._capacity_probe_cycle >= self._capacity_probe_cooldown_until[index]
+                and (weekly_target is not None or index not in self._capacity_probe_attempted)
+                and (
+                    weekly_target is not None
+                    or self._capacity_probe_cycle >= self._capacity_probe_cooldown_until[index]
+                )
                 and self._effective_count_locked(index) >= self._account_caps[index]
                 and (
                     reserve_capacity_slots is None
@@ -815,6 +987,7 @@ class ModelHubClientPool:
                 key=lambda index: (index - self._selection_cursor) % len(self.clients),
             )
             self._selection_cursor = (selected_index + 1) % len(self.clients)
+            if weekly_target is None:
                 self._capacity_probe_attempted.add(selected_index)
             self._reservation_sequence += 1
             reservation_id = self._reservation_sequence
@@ -848,7 +1021,12 @@ class ModelHubClientPool:
                 if observed_capacity < self._account_caps[index]:
                     self._account_caps[index] = observed_capacity
                     cap_changed = True
-            self._remote_counts[index] = self._account_caps[index]
+            # Accepted local reservations may not have appeared in the last
+            # remote count yet. Do not count them twice when marking full.
+            self._remote_counts[index] = max(
+                0,
+                self._account_caps[index] - len(self._reservations[index]),
+            )
             self._count_known[index] = True
             self._counts_initialized = True
             self._active_refresh_at = time.monotonic()
@@ -861,14 +1039,16 @@ class ModelHubClientPool:
             if discovered_cap <= self._account_caps[index]:
                 return
             self._account_caps[index] = discovered_cap
+            weekly = self._weekly_target_locked() == index
         self._persist_account_caps()
+        if not weekly or discovered_cap % 25 == 0:
             print(f"[capacity] account={index + 1:02d} discovered_cap={discovered_cap}", flush=True)
 
     def old_model_submit_slots(self) -> int:
         """Return queue positions where models older than the recent window are allowed."""
         self._refresh_active_counts()
         with self._state_lock:
-            return sum(
+            normal_slots = sum(
                 max(
                     0,
                     self._old_model_count_limit_locked(index) - self._effective_count_locked(index),
@@ -876,6 +1056,7 @@ class ModelHubClientPool:
                 for index in range(len(self.clients))
                 if self._count_known[index]
             )
+            return max(1, normal_slots) if self._weekly_target_locked() is not None else normal_slots
 
     def _old_model_count_limit_locked(self, index: int) -> int:
         return max(0, self._account_caps[index] - self.recent_model_reserve_slots)
@@ -914,9 +1095,15 @@ class ModelHubClientPool:
         attempted_accounts: set[int] = set()
         forced_refresh_done = False
         last_capacity_error: ModelHubAPIError | None = None
-        reserved_capacity_slots = self.recent_model_reserve_slots if old_model_only else None
 
         while True:
+            with self._state_lock:
+                weekly_target = self._weekly_target_locked()
+            reserved_capacity_slots = (
+                self.recent_model_reserve_slots
+                if old_model_only and weekly_target is None
+                else None
+            )
             reservation = self._reserve_account(
                 attempted_accounts,
                 reserve_capacity_slots=reserved_capacity_slots,
@@ -948,6 +1135,9 @@ class ModelHubClientPool:
 
             selected_index, reservation_id = reservation
             selected_client = self.clients[selected_index]
+            if weekly_target == selected_index and time.monotonic() >= self._weekly_calibration_deadline:
+                self._finish_reservation(selected_index, reservation_id, succeeded=False)
+                continue
             try:
                 response = selected_client.add_task(payload)
             except ModelHubAPIError as exc:
@@ -957,6 +1147,18 @@ class ModelHubClientPool:
                 # Another process may have filled this account after our count
                 # refresh. Mark it full locally and immediately try another one.
                 self._mark_account_saturated(selected_index, capacity_probe=capacity_probe)
+                if weekly_target == selected_index:
+                    self._refresh_active_counts(force=True)
+                    with self._state_lock:
+                        confirmed = (
+                            self._count_known[selected_index]
+                            and self._remote_counts[selected_index] >= self._account_caps[selected_index]
+                        )
+                    self.finish_weekly_calibration(
+                        selected_index,
+                        result="saturated" if confirmed else "inconclusive_capacity_response",
+                    )
+                    self._refresh_active_counts(force=True)
                 attempted_accounts.add(selected_index)
                 last_capacity_error = exc
                 continue

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
@@ -218,7 +218,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--capacity-probe-interval-cycles",
         type=int,
-        default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "3")),
+        default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "0")),
         help=argparse.SUPPRESS,
     )
     parser.add_argument("--daily-runs-dir", default=str(DEFAULT_DAILY_RUNS_DIR), help=argparse.SUPPRESS)
@@ -278,7 +278,7 @@ def _build_modelhub_client(base_args: argparse.Namespace) -> ModelHubClientPool:
     clients = [ModelHubClient(token=token, base_url=base_args.modelhub_base_url) for token in token_values]
     return ModelHubClientPool(
         clients,
-        capacity_probe_interval_cycles=max(0, int(getattr(base_args, "capacity_probe_interval_cycles", 3) or 0)),
+        capacity_probe_interval_cycles=0,
         capacity_state_path=Path(getattr(base_args, "capacity_state_path", DEFAULT_CAPACITY_STATE_PATH)),
         recent_model_reserve_slots=max(0, int(getattr(base_args, "recent_model_reserve_slots", 5) or 0)),
         recent_model_days=max(1, int(getattr(base_args, "recent_model_days", 7) or 7)),
@@ -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
 
@@ -857,8 +862,8 @@ def run_poll_loop(
                 log(f"[poll] cycle={cycles} sleep={base_args.idle_interval_seconds}s reason=state_sync_unhealthy")
                 time.sleep(base_args.idle_interval_seconds)
                 continue
-        if hasattr(modelhub_client, "configure_capacity_probe"):
+        if hasattr(modelhub_client, "configure_weekly_calibration"):
-            modelhub_client.configure_capacity_probe(cycles)
+            modelhub_client.configure_weekly_calibration(utc_now())
 
         outcome_synced_this_cycle = False
         if (
@@ -1030,11 +1035,17 @@ def run_poll_loop(
         active_counts = modelhub_client.active_task_counts() if hasattr(modelhub_client, "active_task_counts") else []
         capacity_limits = modelhub_client.account_capacity_limits() if hasattr(modelhub_client, "account_capacity_limits") else []
         capacity_probe = modelhub_client.capacity_probe_enabled() if hasattr(modelhub_client, "capacity_probe_enabled") else False
+        calibration = (
+            modelhub_client.weekly_calibration_status()
+            if hasattr(modelhub_client, "weekly_calibration_status") else {}
+        )
         available_slots = modelhub_client.available_submit_slots() if hasattr(modelhub_client, "available_submit_slots") else None
         log(
             f"[poll] cycle={cycles} active_counts={','.join(str(count) for count in active_counts) if active_counts else 'n/a'} "
             f"capacity_limits={','.join(str(limit) for limit in capacity_limits) if capacity_limits else 'n/a'} "
             f"capacity_probe={'on' if capacity_probe else 'off'} "
+            f"weekly_calibration={'on' if calibration.get('targetAccount') else 'off'} "
+            f"calibration_account={calibration.get('targetAccount') or 'n/a'} "
             f"available_slots={available_slots if available_slots is not None else 'n/a'}"
         )
 
@@ -1074,6 +1085,22 @@ def run_poll_loop(
         cycle_summaries.append(cycle_summary)
         submitted_total += cycle_summary["submittedTotal"]
         remaining_before_run = cycle_summary.get("remainingDailyQuotaBeforeRun")
+        if (
+            calibration.get("targetAccount")
+            and not cycle_summary.get("submittedTotal")
+            and cycle_summary.get("waveResults")
+            and not any(
+                (wave.get("summary") or {}).get("paused")
+                or (wave.get("summary") or {}).get("failedCount", 0)
+                for wave in (cycle_summary.get("waveResults") or [])
+            )
+            and not any(
+                (wave.get("summary") or {}).get("candidateCount", 0)
+                for wave in (cycle_summary.get("waveResults") or [])
+            )
+            and hasattr(modelhub_client, "pause_weekly_calibration_for_no_candidates")
+        ):
+            modelhub_client.pause_weekly_calibration_for_no_candidates()
 
         log(
             f"[poll] cycle_done submitted_total={cycle_summary['submittedTotal']} "

modelhub_submmit_api/version.py

@@ -1 +1 @@
-AGENT_VERSION = "2026.09.18.1"
+AGENT_VERSION = "2026.09.20.2"

tests/test_concurrency.py

@@ -20,6 +20,7 @@ from common import read_json, read_jsonl  # noqa: E402
 from main import build_adaptive_scan_stages, build_parser, make_run_dir, run_submission, submit_candidate  # noqa: E402
 from modelhub_client import (  # noqa: E402
     ModelHubAPIError,
+    ModelHubClient,
     ModelHubClientPool,
     OldModelQueuePolicyError,
     is_duplicate_submission_error,
@@ -120,8 +121,8 @@ class DynamicCapacityClient:
         self.limit = limit
         self.token = token
 
-    def count_active_tasks(self, *, max_count: int, **_kwargs) -> int:  # noqa: ANN003
+    def count_active_tasks(self, *, max_count: int | None, **_kwargs) -> int:  # noqa: ANN003
-        return min(self.active, max_count)
+        return self.active if max_count is None else min(self.active, max_count)
 
     def add_task(self, _payload: dict) -> dict:
         if self.active >= self.limit:
@@ -215,6 +216,195 @@ def make_candidate(index: int) -> dict:
 
 
 class ClientPoolConcurrencyTests(unittest.TestCase):
+    def test_exact_active_count_uses_filtered_platform_totals(self) -> None:
+        client = ModelHubClient(token="test-token")
+        calls: list[str] = []
+
+        def page(*, status: str, **_kwargs) -> dict:  # noqa: ANN003
+            calls.append(status)
+            return {"data": {"total": {"waiting": 175, "running": 25}[status]}}
+
+        client.list_tasks_page = page  # type: ignore[method-assign]
+        self.assertEqual(200, client.count_active_tasks_exact())
+        self.assertEqual(["waiting", "running"], calls)
+
+    @staticmethod
+    def monday_calibration_time(hour: int = 10) -> datetime:
+        # 2026-09-21 10:00 Beijing time.
+        return datetime(2026, 9, 21, hour - 8, tzinfo=timezone.utc)
+
+    def test_weekly_calibration_borrows_five_slots_only_during_monday_window(self) -> None:
+        client = DynamicCapacityClient(active=95, limit=100)
+        pool = ModelHubClientPool(
+            [client],  # type: ignore[list-item]
+            active_task_cap=100,
+            recent_model_reserve_slots=5,
+        )
+        self.assertEqual([95], pool.active_task_counts())
+        old_time = datetime(2026, 9, 1, tzinfo=timezone.utc)
+        pool.configure_weekly_calibration(self.monday_calibration_time(9))
+        with self.assertRaises(OldModelQueuePolicyError):
+            pool.add_task_for_model({"model": "before"}, model_last_modified=old_time)
+
+        pool.configure_weekly_calibration(self.monday_calibration_time())
+        self.assertEqual(1, pool.old_model_submit_slots())
+        for index in range(5):
+            pool.add_task_for_model({"model": index}, model_last_modified=old_time)
+        self.assertEqual(100, client.active)
+        with self.assertRaises(ModelHubAPIError):
+            pool.add_task_for_model({"model": "test-101"}, model_last_modified=old_time)
+        self.assertEqual(1, pool.weekly_calibration_status()["completedAccounts"])
+        self.assertEqual([95], pool.old_model_queue_thresholds())
+        pool.configure_weekly_calibration(self.monday_calibration_time(11))
+        with self.assertRaises(OldModelQueuePolicyError):
+            pool.add_task_for_model({"model": "after"}, model_last_modified=old_time)
+        self.assertEqual(100, client.active)
+
+    def test_weekly_calibration_stops_when_window_deadline_passes(self) -> None:
+        client = DynamicCapacityClient(active=95, limit=200)
+        pool = ModelHubClientPool([client], active_task_cap=100)  # type: ignore[list-item]
+        pool.active_task_counts()
+        pool.configure_weekly_calibration(self.monday_calibration_time())
+        pool._weekly_calibration_deadline = 0.0
+        with self.assertRaises(OldModelQueuePolicyError):
+            pool.add_task_for_model(
+                {"model": "too-late"},
+                model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc),
+            )
+        self.assertEqual(95, client.active)
+
+    def test_weekly_calibration_discovers_200_and_persists_per_account_completion(self) -> None:
+        client = DynamicCapacityClient(active=95, limit=200)
+        old_time = datetime(2026, 9, 1, tzinfo=timezone.utc)
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            state_path = Path(temporary_dir) / "account_capacity.json"
+            pool = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                capacity_state_path=state_path,
+            )
+            pool.active_task_counts()
+            pool.configure_weekly_calibration(self.monday_calibration_time())
+            for index in range(105):
+                pool.add_task_for_model({"model": index}, model_last_modified=old_time)
+            with self.assertRaises(ModelHubAPIError):
+                pool.add_task_for_model({"model": "test-201"}, model_last_modified=old_time)
+            self.assertEqual([200], pool.account_capacity_limits())
+            self.assertEqual([195], pool.old_model_queue_thresholds())
+            stored = read_json(state_path)
+            account = next(iter(stored["accounts"].values()))
+            self.assertEqual("saturated", account["calibrationResult"])
+
+            restarted = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                capacity_state_path=state_path,
+            )
+            restarted.active_task_counts()
+            restarted.configure_weekly_calibration(self.monday_calibration_time())
+            self.assertEqual([200], restarted.account_capacity_limits())
+            self.assertIsNone(restarted.weekly_calibration_status()["targetAccount"])
+
+    def test_weekly_calibration_reads_full_active_count_not_cap_plus_one(self) -> None:
+        client = DynamicCapacityClient(active=200, limit=200)
+        pool = ModelHubClientPool([client], active_task_cap=100)  # type: ignore[list-item]
+        pool.configure_weekly_calibration(self.monday_calibration_time())
+        self.assertEqual([200], pool.active_task_counts())
+        self.assertEqual([200], pool.account_capacity_limits())
+        with self.assertRaises(ModelHubAPIError):
+            pool.add_task_for_model(
+                {"model": "full-capacity"},
+                model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc),
+            )
+        self.assertEqual(1, pool.weekly_calibration_status()["completedAccounts"])
+
+    def test_weekly_calibration_targets_accounts_separately(self) -> None:
+        first = DynamicCapacityClient(active=2, limit=2, token="first")
+        second = DynamicCapacityClient(active=1, limit=3, token="second")
+        pool = ModelHubClientPool(
+            [first, second],  # type: ignore[list-item]
+            active_task_cap=2,
+            recent_model_reserve_slots=1,
+        )
+        pool.active_task_counts()
+        pool.configure_weekly_calibration(self.monday_calibration_time())
+        self.assertEqual(1, pool.weekly_calibration_status()["targetAccount"])
+        pool.add_task_for_model(
+            {"model": "try-first-then-second"},
+            model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc),
+        )
+        self.assertEqual(2, first.active)
+        self.assertEqual(2, second.active)
+        self.assertEqual(2, pool.weekly_calibration_status()["targetAccount"])
+        pool.add_task_for_model(
+            {"model": "second-capacity"},
+            model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc),
+        )
+        with self.assertRaises(ModelHubAPIError):
+            pool.add_task_for_model(
+                {"model": "test-second-full"},
+                model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc),
+            )
+        self.assertEqual([2, 3], pool.account_capacity_limits())
+        self.assertEqual([1, 2], pool.old_model_queue_thresholds())
+        self.assertEqual(2, pool.weekly_calibration_status()["completedAccounts"])
+
+    def test_weekly_calibration_skips_unknown_counts_and_retries_no_candidates(self) -> None:
+        class UnreadableCountClient(DynamicCapacityClient):
+            def count_active_tasks(self, **_kwargs) -> int:
+                raise TimeoutError("cannot read active count")
+
+        unreadable = UnreadableCountClient(active=95, limit=100, token="unreadable")
+        readable = DynamicCapacityClient(active=95, limit=100, token="readable")
+        pool = ModelHubClientPool(
+            [unreadable, readable],  # type: ignore[list-item]
+            active_task_cap=100,
+        )
+        pool.active_task_counts()
+        pool.configure_weekly_calibration(self.monday_calibration_time())
+        self.assertEqual(2, pool.weekly_calibration_status()["targetAccount"])
+        pool.pause_weekly_calibration_for_no_candidates()
+        self.assertEqual(0, pool.weekly_calibration_status()["completedAccounts"])
+        self.assertIsNone(pool.weekly_calibration_status()["targetAccount"])
+        pool._weekly_no_candidate_retry_at = 0.0
+        self.assertEqual(2, pool.weekly_calibration_status()["targetAccount"])
+
+    def test_rate_limit_does_not_finish_weekly_capacity_calibration(self) -> None:
+        class RateLimitedClient(DynamicCapacityClient):
+            def add_task(self, _payload: dict) -> dict:
+                raise ModelHubAPIError("rate limited", code=429)
+
+        client = RateLimitedClient(active=100, limit=200)
+        pool = ModelHubClientPool([client], active_task_cap=100)  # type: ignore[list-item]
+        pool.active_task_counts()
+        pool.configure_weekly_calibration(self.monday_calibration_time())
+        with self.assertRaises(ModelHubAPIError):
+            pool.add_task_for_model(
+                {"model": "rate-limited"},
+                model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc),
+            )
+        self.assertEqual([100], pool.account_capacity_limits())
+        self.assertEqual(1, pool.weekly_calibration_status()["targetAccount"])
+
+    def test_rate_limited_submission_is_deferred_not_failed_or_capacity(self) -> None:
+        class RateLimitedClient:
+            @staticmethod
+            def add_task(_payload: dict) -> dict:
+                raise ModelHubAPIError("rate limited", code=429)
+
+        result = submit_candidate(
+            {
+                "repoId": "owner/model",
+                "modelAddress": "https://modelscope.cn/models/owner/model",
+                "targetGpu": "gpu-a",
+                "framework": "vllm",
+                "taskType": "text-generation",
+                "configParams": "safe",
+            },
+            RateLimitedClient(),  # type: ignore[arg-type]
+        )
+        self.assertEqual("rate_limit_deferred", result["outcome"])
+
     def test_old_models_use_another_account_and_reserve_last_five_positions(self) -> None:
         below_threshold = FakeClient(active_count=94)
         at_threshold = FakeClient(active_count=95)
@@ -544,7 +734,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:

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:

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(

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")

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):