fix: retry failed model inspections and preserve rejection evidence

Do not cache transient repository-list failures as empty weight sets.
Deduplicate normalized file paths without lowering reported file sizes.
Record whitelisted OOM evidence and bounded discovery diagnostics.

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`
@@ -107,8 +107,10 @@ model/GPU uniqueness response, the entire submission cycle pauses fail-closed.
 For each compatible model/GPU pair, the worker also ranks the GPU's supported
 frameworks using ModelHub's public aggregate `modelCount` and `successCount` data,
 then blends in the worker's own GPU+framework outcomes with a capped weight.
-Only frameworks with at least 300 public samples and a safe Wilson lower bound
-are eligible. Framework statistics refresh
+Only frameworks with at least 300 public samples and a Wilson lower bound of at
+least 20% are eligible. GPU pools normally require a 15% lower bound; a
+low-sample pool can remain available only when it has a recent success and a
+framework that passes the stricter framework gate. Framework statistics refresh
 every 6 hours, so they do not add per-model API traffic. Newly published
 frameworks are discovered automatically, but receive no novelty bonus: they can
 win only when their confidence-adjusted success score beats the best incumbent by
@@ -124,7 +126,10 @@ evidence on every scan; accepted route counters and lifetime sufficient statisti
 are stored in `.modelhub_state/routing_intelligence.json`.
 
 Five consecutive local failures open a 12-hour GPU/framework circuit breaker.
-A sub-20% success rate over the latest 20 terminal tasks opens a 6-hour breaker.
+Five or more recent terminal results below 10% success open the same breaker,
+including ambiguous runtime failures that are unsafe to keep feeding. A sub-20%
+attributable success rate over the latest 20 terminal tasks opens a 6-hour
+breaker.
 Platform/infrastructure failures are excluded from long-term compatibility rates
 and model/profile breakers. Three consecutive platform failures on a GPU/framework
 instead open a short 30-minute breaker, so a temporary broken runner or lack of an
@@ -228,13 +233,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
-probing raises an account's known limit, its old-model boundary moves with it.
+Unknown model timestamps are treated as old for new submissions. When weekly
+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
@@ -284,11 +301,43 @@ model/GPU combinations, the same run progressively expands discovery to the last
 7 days, the last 30 days, and finally older history (up to 3,000 models). Scanning
 stops as soon as enough replacement candidates have been found.
 
+When only recent-model reserved positions remain, discovery also deepens the
+same recent window up to the configured scan cap (3,000 by default). It never
+admits older models into those positions or lowers the quality gates. This avoids
+rechecking only an exhausted short prefix on every poll.
+
+An ordinary account-capacity rejection temporarily marks that account full and
+tries the next account. It does not lower the persisted capacity using a cached
+active count: another worker may have filled the queue since that count was
+read. Downward capacity learning remains confined to weekly calibration.
+
+Account allocation is age-aware: recent models first use accounts with only
+reserved positions left, then the account with the fewest shared positions
+remaining. Old models continue to use the account with the most eligible space.
+Largest-headroom and rotating tie-breaks balance otherwise equivalent accounts;
+weekly calibration remains pinned to its target account. For example, with
+accounts at 94/100 and 95/100 and five recent-only positions each, assigning a
+recent model to the second account preserves the first account's last old-model
+position. A recent-then-old pair can both be accepted rather than stranding the
+old model after the recent model takes its only eligible position.
+
 Model verification results are reused for 15 minutes across poll cycles, and a
 locally failed model/GPU pair cools down for 24 hours instead of being excluded
 forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done`
 includes `skip_reasons` so an empty candidate pool is directly diagnosable.
 
+Repository-tree HTTP failures are reported as retriable inspection failures,
+not cached as empty repositories. Repeated normalized file paths count once
+(using the largest reported size for that path), while different weight formats
+still contribute separately to the full-repository OOM gate.
+
+For an empty-candidate investigation, read `daily_runs/<cycle>/summary.json`,
+then follow `waveResults[].summary.runDir` to `skipped.jsonl` and `failed.jsonl`.
+OOM skips now include size/capacity/source evidence, and `scanStages` includes
+`discoveryQueries` with raw rows, accepted rows, cache hits, and stop reasons.
+Reaching a page/result budget does not mean the catalog is exhausted. These
+diagnostics contain selected fields only, never full configs or HTTP headers.
+
 Community deduplication is scoped to the exact model/GPU combination. A model
 adapted on one GPU remains eligible for another GPU. Before each submission the
 worker performs an uncached community check; a lookup failure defers the task
@@ -451,11 +500,41 @@ 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.
+
+Version `2026.09.22.1` restores accuracy-first admission while retaining bounded
+capacity recovery. Any existing result for an exact model/GPU pair, including a
+failed result, closes that pair; a different framework no longer reopens it.
+Each model is limited to six GPU routes across poll cycles and at most two local
+routes may await feedback simultaneously. Lower-ranked GPUs are still checked,
+in score order, only when the preferred six cannot produce a safe candidate.
+Download-ranked popularity fallback is removed: discovery expands by update
+window instead. Public framework/GPU confidence floors and an all-terminal
+failure circuit prevent high-volume low-quality routes from winning on relative
+rank alone. Community-confirmed model successes are used only as a tie-breaker
+inside the same confidence band.
+
 ## 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-v36 -m "ModelHub agent 2026.09.20.2"
+git push origin main agent-v36
 ```

modelhub_submmit_api/README.md

@@ -79,21 +79,25 @@ bash run_poll.sh --dry-run
 ## Behavior
 
 - The runner auto-discovers all safe GPU/template combinations from the public submit catalog.
-- Automatic GPU selection uses exact 70/30 accepted-task scheduling: long-term
-  Wilson-ranked top 3 GPUs and the top GPUs from the latest 1,000 terminal tasks.
-  There is no all-GPU exploration category.
-- Within each category, weighted-fair scheduling uses estimated queue backlog hours,
+- Automatic GPU selection uses one success-first score built from qualified
+  local cohorts and ModelHub public evidence. There is no random or fixed-quota
+  exploration category.
+- Within the qualified success band, scheduling uses estimated queue backlog hours,
   recent public throughput/success, machine availability, and worker concurrency.
   Unavailable or stalled GPU pools are circuit-broken instead of continuing to absorb work.
 - Compatible frameworks are ranked by ModelHub public aggregate success statistics
   plus capped local GPU+framework evidence, with a 300-sample public minimum and
-  Wilson confidence bounds. Missing or undersized public evidence receives zero
-  traffic rather than falling back to exploration.
+  a 20% Wilson lower-bound floor. GPU pools normally require a 15% floor.
+  Missing or undersized public evidence receives zero traffic rather than
+  falling back to exploration; a low-sample GPU requires a recent success and a
+  framework that already passes the stricter framework gate.
 - New frameworks are discovered from the live catalog but get no novelty bonus.
   They are eligible only with a complete official build config that passes local
   validation and a confidence score at least 10% above the best incumbent.
-- Five consecutive local failures pause a GPU/framework pair for 12 hours; a
-  sub-20% rate over the latest 20 terminal tasks pauses it for 6 hours.
+- Five consecutive local failures pause a GPU/framework pair for 12 hours. Five
+  or more recent terminal outcomes below 10% success also pause it for 12 hours,
+  including ambiguous runtime failures. A sub-20% attributable rate over the
+  latest 20 terminal tasks pauses it for 6 hours.
 - An explicit "framework does not support this model/architecture" failure learns
   a 30-day GPU + framework + task + architecture block. Architecture identity
   comes from candidate `config.json` plus exact unsupported `model_type` or
@@ -114,23 +118,33 @@ bash run_poll.sh --dry-run
   duplicates do not advance it. The next cycle refreshes platform history before submitting again.
 - 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.
+- Candidate discovery starts with the configured recent window, then
+  automatically expands to 7 days, 30 days, and the first 3,000 raw ModelScope
+  rows ordered by update time. Download popularity is not treated as ModelHub
+  compatibility evidence.
 - 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.
+- Community deduplication is model/GPU-specific: another GPU's adaptation does
+  not block the current GPU. Only `未验证` with no records is eligible;
+  `验证失败`, `已验证`, `验证中`, and ambiguous records remain blocked. Every actual
+  submission performs a fresh uncached check for its exact GPU.
 - 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.
+- A model may use at most six GPU routes across poll cycles, and at most two
+  locally submitted routes for that model may await feedback at once. The first
+  six ranked GPUs remain preferred; lower-ranked eligible GPUs are inspected in
+  order only when that tranche yields no safe candidate.
 - 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
-  raises that account's persisted known limit; a capacity rejection enters cooldown.
+- Each Monday 10:00-10:59 Asia/Shanghai, the worker calibrates accounts one at a time.
+  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 +153,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/candidate_preflight.py

@@ -191,6 +191,8 @@ class CandidatePreflightAdvisor:
             "quantizationMethod": inspection.quantization_method,
             "estimatedLoadBytes": inspection.estimated_load_bytes(framework),
             "repositoryOnDiskBytes": inspection.repository_size_bytes,
+            "recursiveRepositoryBytes": inspection.exact_repository_size_bytes,
+            "publishedSizeBytes": inspection.published_size_bytes,
             "gpuMemoryGiB": self.gpu_memory_gib.get(target_gpu),
             "gpuMemoryEvidence": self.gpu_memory_evidence.get(target_gpu),
         }
@@ -272,7 +274,9 @@ class CandidatePreflightAdvisor:
             required_gib = memory_sizing_bytes / (1024**3) * MODEL_LOAD_OVERHEAD
             metadata["estimatedRequiredGiB"] = round(required_gib, 3)
             metadata["memorySizingBasis"] = (
-                "recursive_repository_on_disk" if repository_bytes else "selected_weights_fallback"
+                "published_repository_upper_bound"
+                if (inspection.published_size_bytes or 0) > (inspection.exact_repository_size_bytes or 0)
+                else ("recursive_repository_on_disk" if repository_bytes else "selected_weights_fallback")
             )
             metadata["maximumRepositorySizeGiB"] = round(memory_gib / MODEL_LOAD_OVERHEAD, 3)
             if required_gib > memory_gib:

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,9 @@ 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
+        self.last_listing_diagnostics: list[dict[str, Any]] = []
 
     @staticmethod
     def _remember_lru(cache: OrderedDict, key: Any, value: Any, limit: int) -> None:
@@ -124,14 +127,28 @@ 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 []
 
         deduped: dict[str, HFModelSummary] = {}
+        self.last_listing_diagnostics = []
         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 +156,9 @@ class HuggingFaceDiscovery:
                 limit=per_tag_limit,
                 min_downloads=min_downloads,
                 updated_after=updated_after,
+                sort=sort,
+                start_page=start_page,
+                diagnostics=self.last_listing_diagnostics,
             ):
                 current = deduped.get(model.repo_id)
                 if current is None or (model.last_modified or parse_datetime("1970-01-01")) > (
@@ -156,6 +176,9 @@ class HuggingFaceDiscovery:
         limit: int,
         min_downloads: int,
         updated_after=None,
+        sort: str = "last_modified",
+        start_page: int = 1,
+        diagnostics: list[dict[str, Any]] | None = None,
     ) -> list[HFModelSummary]:
         # ModelScope OpenAPI rejects values above 50 with InputParameterError.
         page_size = min(max(1, limit), 50)
@@ -163,12 +186,24 @@ 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
+        diagnostic = {
+            "pipelineTag": pipeline_tag, "sort": sort,
+            "requestedLimit": limit, "effectiveLimit": max_items,
+            "pageSize": page_size, "startPage": start_page,
+            "pagesRead": 0, "cachedPages": 0, "rawRowsScanned": 0,
+            "acceptedModels": 0, "stopReason": "page_budget",
+            "catalogExhausted": False,
+        }
+        if diagnostics is not None:
+            diagnostics.append(diagnostic)
+        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)
                 items = list(cached[1])
+                diagnostic["cachedPages"] += 1
             else:
                 elapsed = time.monotonic() - self._last_model_page_request_at
                 if self._last_model_page_request_at > 0 and elapsed < self._page_interval_seconds:
@@ -177,7 +212,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
@@ -191,6 +226,7 @@ class HuggingFaceDiscovery:
                     self._last_model_page_request_at = time.monotonic()
                     if exc.status_code == 400 and not filter_disabled:
                         self._unsupported_task_filters[task_tag] = time.monotonic() + 86_400
+                        diagnostic["stopReason"] = "unsupported_filter_retry"
                         print(
                             f"[modelscope] task_filter_unsupported task={task_tag} "
                             "fallback=unfiltered ttl=86400s",
@@ -201,12 +237,17 @@ class HuggingFaceDiscovery:
                             limit=limit,
                             min_downloads=min_downloads,
                             updated_after=updated_after,
+                            sort=sort,
+                            start_page=start_page,
+                            diagnostics=diagnostics,
                         )
                     print(
                         f"[modelscope] list_models_error task={task_tag} page={page_number} "
                         f"partial_models={len(models)} retry_next_cycle=true error={exc}",
                         flush=True,
                     )
+                    diagnostic["stopReason"] = "http_error"
+                    diagnostic["httpStatus"] = exc.status_code
                     break
 
                 items = self._extract_models(payload)
@@ -216,7 +257,11 @@ class HuggingFaceDiscovery:
                     (time.monotonic(), list(items)),
                     self._page_cache_limit,
                 )
+            diagnostic["pagesRead"] += 1
+            diagnostic["rawRowsScanned"] += len(items)
             if not items:
+                diagnostic["stopReason"] = "empty_page"
+                diagnostic["catalogExhausted"] = True
                 break
             for item in items:
                 model = self._parse_model(
@@ -227,9 +272,13 @@ class HuggingFaceDiscovery:
                 )
                 if model is not None:
                     models.append(model)
+                    diagnostic["acceptedModels"] += 1
                 if len(models) >= max_items:
+                    diagnostic["stopReason"] = "result_limit"
                     return models
             if len(items) < page_size:
+                diagnostic["stopReason"] = "short_page"
+                diagnostic["catalogExhausted"] = True
                 break
         return models
 
@@ -390,7 +439,10 @@ class HuggingFaceDiscovery:
             )
         except HttpJsonError as exc:
             print(f"[modelscope] repo_tree_error repo={repo_id} error={exc}", flush=True)
-            payload = None
+            # A failed lookup is not an empty repository. Let the caller defer
+            # inspection and retry next cycle instead of caching missing
+            # weights as a permanent framework-incompatibility decision.
+            raise
 
         entries = self._extract_files(payload)
 
@@ -548,7 +600,7 @@ def inspect_repo_tree(repo_id: str, entries: list[dict[str, Any]]) -> ModelInspe
                 break
         if size_present:
             try:
-                file_sizes[path] = max(0, int(size_value))
+                file_sizes[path] = max(file_sizes.get(path, 0), int(size_value))
             except (TypeError, ValueError):
                 pass
         filename = PurePosixPath(path).name.lower()
@@ -562,12 +614,12 @@ def inspect_repo_tree(repo_id: str, entries: list[dict[str, Any]]) -> ModelInspe
     selected_gguf = choose_best_gguf(gguf_files)
     return ModelInspection(
         repo_id=repo_id,
-        file_paths=sorted(file_paths),
+        file_paths=sorted(set(file_paths)),
         file_sizes=file_sizes,
-        gguf_files=sorted(gguf_files),
+        gguf_files=sorted(set(gguf_files)),
         selected_gguf=PurePosixPath(selected_gguf).name if selected_gguf else None,
-        weight_files=sorted(vllm_weight_files),
-        onnx_files=sorted(onnx_files),
+        weight_files=sorted(set(vllm_weight_files)),
+        onnx_files=sorted(set(onnx_files)),
     )
 
 

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
@@ -66,6 +66,7 @@ DEFAULT_LEDGER_PATH = Path("ledger/submissions.jsonl")
 ADAPTIVE_SCAN_MAX_MODELS = 3000
 ADAPTIVE_SCAN_MIN_FALLBACK_MODELS = 500
 MAX_ROUTES_PER_MODEL = 6
+MAX_PENDING_ROUTES_PER_MODEL = 2
 
 
 def build_parser() -> argparse.ArgumentParser:
@@ -258,7 +259,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 +376,8 @@ 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,
+    rejection_details: list[dict[str, Any]] | None = None,
 ) -> tuple[CandidateModel | None, str | None]:
     last_rejection_reason: str | None = None
     for task_type in task_types:
@@ -397,6 +400,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
 
@@ -463,6 +474,23 @@ def choose_candidate_for_gpu_detailed(
                 )
                 if not assessment.allowed:
                     last_rejection_reason = assessment.reason
+                    if rejection_details is not None:
+                        rejection_details.append({
+                            "taskType": task_type,
+                            "framework": framework,
+                            "reason": assessment.reason,
+                            "metadata": {
+                                key: assessment.metadata[key]
+                                for key in (
+                                    "modelType", "architectures", "quantizationMethod",
+                                    "estimatedLoadBytes", "repositoryOnDiskBytes",
+                                    "recursiveRepositoryBytes", "publishedSizeBytes",
+                                    "gpuMemoryGiB", "gpuMemoryEvidence",
+                                    "estimatedRequiredGiB", "memorySizingBasis",
+                                    "maximumRepositorySizeGiB",
+                                ) if key in assessment.metadata
+                            },
+                        })
                     continue
                 config_params = assessment.config_params
                 warnings.extend(assessment.warnings)
@@ -508,6 +536,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,14 +591,43 @@ 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())
+            successful_gpus = set(precheck.get("successfulGpus") or set())
+            failed_frameworks_by_gpu = precheck.get("failedFrameworksByGpu") or {}
         else:
             processed_gpus = modelhub_client.processed_gpus_for_model(model.repo_id)
+            successful_gpus = set()
+            failed_frameworks_by_gpu = {}
     except Exception as exc:
         return (
             [],
             [{"repoId": model.repo_id, "reason": "community_precheck_unavailable_fail_closed"}],
             [{"repoId": model.repo_id, "reason": f"community_precheck_error:{exc}"}],
         )
+    if len(processed_gpus) >= MAX_ROUTES_PER_MODEL:
+        return (
+            [],
+            [{
+                "repoId": model.repo_id,
+                "reason": "model_route_limit_reached",
+                "processedGpuCount": len(processed_gpus),
+            }],
+            [],
+        )
+    pending_route_count = (
+        outcome_tracker.pending_route_count_for_model(model.repo_id)
+        if outcome_tracker is not None
+        else 0
+    )
+    if pending_route_count >= MAX_PENDING_ROUTES_PER_MODEL:
+        return (
+            [],
+            [{
+                "repoId": model.repo_id,
+                "reason": "model_pending_feedback_limit",
+                "pendingRouteCount": pending_route_count,
+            }],
+            [],
+        )
     candidates: list[dict[str, Any]] = []
     skipped: list[dict[str, Any]] = []
     failed: list[dict[str, Any]] = []
@@ -580,8 +640,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 +651,34 @@ 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:
+                skipped.append({
+                    "repoId": model.repo_id,
+                    "targetGpu": target_gpu,
+                    "reason": "prior_failed_gpu_closed_accuracy_first",
+                })
+                continue
+            failed_frameworks = set(failed_frameworks_by_gpu.get(target_gpu) or set())
             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",
+                        "preflightDetails": {
+                            "stage": "published_size_precheck",
+                            "publishedSizeBytes": model.file_size,
+                            "estimatedRequiredGiB": model.file_size / (1024**3) * MODEL_LOAD_OVERHEAD,
+                            "gpuMemoryGiB": memory_gib,
+                            "gpuMemoryEvidence": preflight_advisor.gpu_memory_evidence.get(target_gpu),
+                            "loadOverhead": MODEL_LOAD_OVERHEAD,
+                        },
+                    })
+                    continue
 
             exact_task_types = list(allowed_task_types_set)
             if official_registry is not None:
@@ -632,7 +717,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,11 +738,12 @@ 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
+            rejection_details: list[dict[str, Any]] = []
             best, preflight_reason = choose_candidate_for_gpu_detailed(
                 model=model,
                 inspection=inspection,
@@ -664,6 +754,8 @@ 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,
+                rejection_details=rejection_details,
             )
             if best is None:
                 reason = (
@@ -674,9 +766,17 @@ def process_model_for_candidates(
                         else "no_compatible_auto_template_or_framework"
                     )
                 )
-                skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason})
+                skipped.append({
+                    "repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason,
+                    **({"preflightRejections": rejection_details} if rejection_details else {}),
+                })
                 continue
             record = candidate_to_record(best)
+            if excluded_frameworks:
+                record["previousFailedFrameworks"] = sorted(excluded_frameworks)
+            record["communitySuccessfulGpuCount"] = len(successful_gpus)
+            if successful_gpus:
+                record["communitySuccessfulGpus"] = sorted(successful_gpus)
             if outcome_tracker is not None:
                 record["transformersPrerequisiteSatisfied"] = (
                     outcome_tracker.has_non_transformers_failure(model.repo_id)
@@ -753,12 +853,26 @@ def build_adaptive_scan_stages(
                 "limit": max_models,
             }
         )
-
+    if not allow_older_than_recent_window:
+        # Reserved positions still need a deep search of eligible recent
+        # models. Widening the date window alone repeatedly scans the same
+        # short prefix (often already processed) on every poll cycle.
+        stages.append(
+            {
+                "name": "recent_window_full",
+                "updatedAfter": recent_cutoff,
+                "limit": max_models,
+            }
+        )
     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 +991,21 @@ def submit_candidate(
                 "candidate": candidate,
                 "reason": "already_processed_for_gpu",
             }
+        if len(set(precheck.get("processedGpus") or set())) >= MAX_ROUTES_PER_MODEL:
+            return {
+                "outcome": "duplicate",
+                "candidate": candidate,
+                "reason": "model_route_limit_reached",
+            }
+        failed_frameworks = set(
+            (precheck.get("failedFrameworksByGpu") or {}).get(candidate["targetGpu"]) or set()
+        )
+        if failed_frameworks:
+            return {
+                "outcome": "duplicate",
+                "candidate": candidate,
+                "reason": "prior_failed_gpu_closed_accuracy_first",
+            }
 
     payload = {
         "modelAddress": candidate["modelAddress"],
@@ -964,6 +1093,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 +1204,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 +1554,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 +1589,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 +1652,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),
@@ -1510,6 +1662,7 @@ def run_submission(
             "candidateModelsAfterStage": candidate_model_count(candidates),
             "skippedAdded": len(skipped) - skipped_before_stage,
             "failedAdded": len(stage_failed),
+            "discoveryQueries": list(getattr(hf_discovery, "last_listing_diagnostics", [])),
         }
         scan_stages.append(stage_summary)
         print(
@@ -1648,10 +1801,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 +1968,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

@@ -20,7 +20,8 @@ DEFAULT_THROUGHPUT_WINDOW_HOURS = 6
 DEFAULT_FETCH_WORKERS = 4
 DEFAULT_FRAMEWORK_MIN_SAMPLES = 300
 DEFAULT_GPU_MIN_RECENT_TERMINALS = 20
-DEFAULT_FRAMEWORK_MIN_WILSON = 0.05
+DEFAULT_FRAMEWORK_MIN_WILSON = 0.20
+DEFAULT_GPU_MIN_WILSON = 0.15
 DEFAULT_COMMUNITY_WINDOW_DAYS = 30
 DEFAULT_COMMUNITY_MAX_MODELS = 200
 DEFAULT_COMMUNITY_HYDRATE_PER_REFRESH = 50
@@ -201,7 +202,7 @@ class MarketIntelligenceManager:
                 and health_factor >= 0.5
                 and int(item.get("recentTerminal") or 0) >= DEFAULT_GPU_MIN_RECENT_TERMINALS
                 and int(item.get("recentSuccess") or 0) > 0
-                and wilson >= DEFAULT_FRAMEWORK_MIN_WILSON
+                and wilson >= DEFAULT_GPU_MIN_WILSON
             )
 
     def _base_state(self, gpus: list[str], task_types: list[str], now: datetime) -> dict[str, Any]:
@@ -554,7 +555,7 @@ class MarketIntelligenceManager:
                 and health_factor >= 0.5
                 and int(stats.get("recentTerminal") or 0) >= DEFAULT_GPU_MIN_RECENT_TERMINALS
                 and int(stats.get("recentSuccess") or 0) > 0
-                and wilson >= DEFAULT_FRAMEWORK_MIN_WILSON
+                and wilson >= DEFAULT_GPU_MIN_WILSON
             )
             result[gpu] = stats
 
@@ -661,7 +662,26 @@ class MarketIntelligenceManager:
             # Missing live data falls back to the scheduler's already-proven
             # long/recent pools. Explicitly failed public gates never do.
             if eligibility is False:
-                continue
+                low_sample = int(item.get("recentTerminal") or 0) < DEFAULT_GPU_MIN_RECENT_TERMINALS
+                has_recent_success = int(item.get("recentSuccess") or 0) > 0
+                recent_wilson = float(item.get("recentWilsonLowerBound") or 0.0)
+                healthy = float(item.get("healthFactor") or 1.0) >= 0.5
+                has_proven_framework = any(
+                    bool(framework.get("officialConfigValid", False))
+                    and int(framework.get("modelCount") or 0) >= self.framework_min_samples
+                    and float(framework.get("wilsonLowerBound") or 0.0) >= DEFAULT_FRAMEWORK_MIN_WILSON
+                    for by_gpu in ((self.state or {}).get("frameworkStats") or {}).values()
+                    for framework in ((by_gpu or {}).get(gpu) or {}).values()
+                )
+                if (
+                    item.get("canVerify") is False
+                    or not low_sample
+                    or not has_recent_success
+                    or recent_wilson < DEFAULT_GPU_MIN_WILSON
+                    or not healthy
+                    or not has_proven_framework
+                ):
+                    continue
             ranked.append((-float(item.get("selectionWeight") or 1.0), index, gpu))
         return [gpu for _weight, _index, gpu in sorted(ranked)]
 
@@ -723,6 +743,7 @@ class MarketIntelligenceManager:
 
         recent_item = (self.local_outcome_stats.get("recentCombinationStats") or {}).get(local_key) or {}
         recent_success = max(0, int(recent_item.get("successCount") or 0))
+        recent_all_failures = max(0, int(recent_item.get("failureCount") or 0))
         recent_failure = max(
             0,
             int(recent_item.get("attributableFailureCount", recent_item.get("failureCount") or 0)),
@@ -743,6 +764,16 @@ class MarketIntelligenceManager:
         elif last_terminal_at is not None and consecutive_failures >= 5:
             circuit_reason = "five_consecutive_local_failures"
             circuit_until = last_terminal_at + timedelta(hours=12)
+        elif (
+            last_terminal_at is not None
+            and recent_success + recent_all_failures >= 5
+            and recent_success / (recent_success + recent_all_failures) < 0.10
+        ):
+            # Runtime-empty and other unresolved failures must not create a
+            # permanent compatibility blacklist. Repetition is still strong
+            # operational evidence, so pause the route while feedback settles.
+            circuit_reason = "recent_terminal_success_below_10_percent"
+            circuit_until = last_terminal_at + timedelta(hours=12)
         elif last_terminal_at is not None and recent_samples >= 20 and recent_rate is not None and recent_rate < 0.20:
             circuit_reason = "recent_local_success_below_20_percent"
             circuit_until = last_terminal_at + timedelta(hours=6)
@@ -859,6 +890,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,14 +37,59 @@ 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()
+    successful_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
+        blocked_gpus.add(name)
+        if status == "已验证" or any(
+            isinstance(record, dict) and _positive_verify_result(record.get("verifyResult"))
+            for record in (records or [])
+        ):
+            successful_gpus.add(name)
+        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
+        # Accuracy-first admission: any existing model/GPU state remains
+        # closed. Failed-framework details are diagnostic evidence only.
+
     return {
         "isInDB": data.get("isInDB") is True,
-        "processedGpus": set(str(gpu) for gpu in verify_result),
+        "processedGpus": blocked_gpus,
+        "successfulGpus": successful_gpus,
+        "failedFrameworksByGpu": failed_frameworks_by_gpu,
         "verifyResult": verify_result,
         "modelInfo": data.get("modelInfo") if isinstance(data.get("modelInfo"), dict) else {},
     }
 
 
+def _positive_verify_result(value: Any) -> bool:
+    try:
+        return int(value) > 0
+    except (TypeError, ValueError):
+        return False
+
+
 class ModelHubClient:
     def __init__(
         self,
@@ -73,10 +119,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)
@@ -281,6 +324,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 +434,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 +470,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 +495,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 +550,15 @@ class ModelHubClientPool:
                     stored = payload.get("accounts") or {}
             except (FileNotFoundError, ValueError):
                 pass
-        return [
-            max(1, int((stored.get(key) or {}).get("knownCap") or default_cap))
-            for key in self._account_keys
-        ]
+        caps: list[int] = []
+        for index, key in enumerate(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 +568,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 +591,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 +701,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 +733,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 +839,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 +930,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 (
@@ -768,6 +948,26 @@ class ModelHubClientPool:
             usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0}
             if not usable:
                 return None
+            if (
+                reserve_capacity_slots is None
+                and weekly_target is None
+                and self.recent_model_reserve_slots > 0
+            ):
+                # Recent models can use every position; old models cannot.
+                # Pack recent work into reserved positions first, then the
+                # account nearest its old-model threshold. This preserves
+                # shared positions for later old candidates, regardless of
+                # candidate arrival order. Keep largest-headroom/round-robin
+                # balancing among accounts with the same shared capacity.
+                shared_slots = {
+                    index: max(0, remaining - self.recent_model_reserve_slots)
+                    for index, remaining in usable.items()
+                }
+                least_shared = min(shared_slots.values())
+                usable = {
+                    index: remaining for index, remaining in usable.items()
+                    if shared_slots[index] == least_shared
+                }
             best_remaining = max(usable.values())
             tied = [index for index, remaining in usable.items() if remaining == best_remaining]
             selected_index = min(
@@ -790,14 +990,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 self._capacity_probe_cycle >= self._capacity_probe_cooldown_until[index]
+                and (weekly_target is None or index == weekly_target)
+                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,7 +1020,8 @@ class ModelHubClientPool:
                 key=lambda index: (index - self._selection_cursor) % len(self.clients),
             )
             self._selection_cursor = (selected_index + 1) % len(self.clients)
-            self._capacity_probe_attempted.add(selected_index)
+            if weekly_target is None:
+                self._capacity_probe_attempted.add(selected_index)
             self._reservation_sequence += 1
             reservation_id = self._reservation_sequence
             self._reservations[selected_index][reservation_id] = {
@@ -843,12 +1049,22 @@ class ModelHubClientPool:
                 self._capacity_probe_cooldown_until[index] = (
                     self._capacity_probe_cycle + self._capacity_probe_cooldown_cycles
                 )
-            else:
+            elif self._weekly_target_locked() == index:
+                # Only the scheduled calibration may lower a learned cap.
+                # Ordinary rejection can race with another agent filling this
+                # account; our cached count is not evidence of its capacity.
+                # Persisting it (even zero -> one) strands otherwise usable
+                # positions until the next calibration.
                 observed_capacity = max(1, self._effective_count_locked(index))
                 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 +1077,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()
-        print(f"[capacity] account={index + 1:02d} discovered_cap={discovered_cap}", flush=True)
+        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 +1094,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 +1133,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 +1173,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 +1185,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/models.py

@@ -140,12 +140,17 @@ class ModelInspection:
         return None
 
     @property
-    def repository_size_bytes(self) -> int | None:
+    def exact_repository_size_bytes(self) -> int | None:
         """Return exact recursive on-disk size when every file has a size."""
-        exact: int | None = None
         if self.file_paths and not any(path not in self.file_sizes for path in self.file_paths):
             total = sum(max(0, int(self.file_sizes[path])) for path in self.file_paths)
-            exact = total if total > 0 else None
+            return total if total > 0 else None
+        return None
+
+    @property
+    def repository_size_bytes(self) -> int | None:
+        """Use the conservative maximum of recursive and published sizes."""
+        exact = self.exact_repository_size_bytes
         published = int(self.published_size_bytes or 0) or None
         values = [value for value in (exact, published) if value is not None]
         return max(values) if values else None

modelhub_submmit_api/outcome_tracker.py

@@ -344,6 +344,17 @@ class OutcomeTracker:
         self._by_model_gpu[(model_id, target_gpu)].append(record)
         append_jsonl(self.path, record)
 
+    def pending_route_count_for_model(self, model_id: str) -> int:
+        """Return distinct locally submitted GPU routes still awaiting an outcome."""
+        pending_gpus: set[str] = set()
+        for record in self._records:
+            if record.get("modelId") != model_id or record.get("outcome") != "pending":
+                continue
+            target_gpu = str(record.get("targetGpu") or "").strip()
+            if target_gpu:
+                pending_gpus.add(target_gpu)
+        return len(pending_gpus)
+
     def mark_policy_cancellations(self, decisions: list[dict[str, Any]]) -> int:
         """Persist intentional task stops so they never become failure evidence."""
         marked = 0

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"):
-            modelhub_client.configure_capacity_probe(cycles)
+        if hasattr(modelhub_client, "configure_weekly_calibration"):
+            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/routing_engine.py

@@ -123,6 +123,9 @@ class SuccessFirstRoutingEngine:
                 "routingEtaHours": eta,
                 "routingEtaUnknown": eta_unknown,
                 "routingExpectedSuccessPerHour": expected,
+                "routingCommunityModelSuccesses": max(
+                    0, int(candidate.get("communitySuccessfulGpuCount") or 0)
+                ),
                 "strategyCategory": "unified_success_first",
             }
         )
@@ -168,6 +171,7 @@ class SuccessFirstRoutingEngine:
         selected.sort(
             key=lambda item: (
                 -int(item["routingSuccessBand"]),
+                -int(item.get("routingCommunityModelSuccesses") or 0),
                 -float(item["routingExpectedSuccessPerHour"]),
                 -int(item["routingSamples"]),
                 -int(item.get("downloads") or 0),

modelhub_submmit_api/version.py

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

tests/test_candidate_preflight.py

@@ -243,6 +243,26 @@ class CandidatePreflightTests(unittest.TestCase):
         self.assertFalse(assessment.allowed)
         self.assertEqual("recursive_repository_on_disk", assessment.metadata["memorySizingBasis"])
 
+    def test_memory_diagnostics_distinguish_published_upper_bound(self) -> None:
+        gib = 1024 ** 3
+        inspection = ModelInspection(
+            repo_id="owner/published-larger",
+            file_paths=["config.json", "tokenizer.json", "model.safetensors"],
+            file_sizes={"config.json": 0, "tokenizer.json": 0, "model.safetensors": 20 * gib},
+            weight_files=["model.safetensors"],
+            model_config={"model_type": "qwen2"},
+            published_size_bytes=40 * gib,
+        )
+        assessment = CandidatePreflightAdvisor(gpu_memory_gib={"gpu": 32}).assess(
+            inspection=inspection, task_type="text-generation", target_gpu="gpu",
+            framework="vllm", config_params="max_model_len: 4096",
+        )
+        self.assertEqual("preflight_predicted_oom", assessment.reason)
+        self.assertEqual(20 * gib, assessment.metadata["recursiveRepositoryBytes"])
+        self.assertEqual(40 * gib, assessment.metadata["publishedSizeBytes"])
+        self.assertEqual(48.0, assessment.metadata["estimatedRequiredGiB"])
+        self.assertEqual("published_repository_upper_bound", assessment.metadata["memorySizingBasis"])
+
     def test_p800_verified_capacity_accepts_80_gib_repository_boundary(self) -> None:
         gib = 1024**3
 

tests/test_concurrency.py

@@ -6,6 +6,7 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from unittest.mock import patch
@@ -20,6 +21,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 +122,8 @@ class DynamicCapacityClient:
         self.limit = limit
         self.token = token
 
-    def count_active_tasks(self, *, max_count: int, **_kwargs) -> int:  # noqa: ANN003
-        return min(self.active, max_count)
+    def count_active_tasks(self, *, max_count: int | None, **_kwargs) -> int:  # noqa: ANN003
+        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 +217,264 @@ def make_candidate(index: int) -> dict:
 
 
 class ClientPoolConcurrencyTests(unittest.TestCase):
+    def test_recent_work_preserves_shared_slots_for_later_old_models(self) -> None:
+        for initial_counts, ages, expected_counts in (
+            ([94, 95], ["recent", "old"], [95, 96]),
+            ([93, 94], ["recent", "recent", "old", "old"], [95, 96]),
+        ):
+            with self.subTest(initial_counts=initial_counts):
+                clients = [FakeClient(active_count=count) for count in initial_counts]
+                pool = ModelHubClientPool(
+                    clients,  # type: ignore[arg-type]
+                    active_task_cap=100,
+                    recent_model_reserve_slots=5,
+                    active_counts_ttl=60,
+                    capacity_probe_interval_cycles=0,
+                )
+                for index, age in enumerate(ages):
+                    pool.add_task_for_model(
+                        {"model": f"model-{index}"},
+                        model_last_modified=(
+                            "2026-10-01T00:00:00Z" if age == "recent"
+                            else "2025-01-01T00:00:00Z"
+                        ),
+                        submitted_at="2026-10-02T00:00:00Z",
+                    )
+                self.assertEqual(expected_counts, pool.active_task_counts())
+                self.assertEqual(len(ages), sum(len(client.submitted) for client in clients))
+                self.assertEqual(0, pool.old_model_submit_slots())
+
+    def test_recent_packing_keeps_largest_headroom_without_reserved_slots(self) -> None:
+        clients = [FakeClient(active_count=2), FakeClient(active_count=3)]
+        pool = ModelHubClientPool(
+            clients,  # type: ignore[arg-type]
+            active_task_cap=5,
+            recent_model_reserve_slots=0,
+        )
+        pool.add_task({"model": "recent"})
+        self.assertEqual([1, 0], [len(client.submitted) for client in clients])
+
+    def test_stale_capacity_rejection_does_not_poison_learned_limit(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            state_path = Path(temporary_dir) / "capacity.json"
+            client = FakeClient(active_count=0, reject_capacity=True)
+            pool = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                recent_model_reserve_slots=5,
+                capacity_probe_interval_cycles=0,
+                capacity_state_path=state_path,
+            )
+            self.assertEqual(100, pool.available_submit_slots())
+            with self.assertRaises(ModelHubAPIError):
+                pool.add_task({"model": "rejected-during-race"})
+
+            self.assertEqual([100], pool.account_capacity_limits())
+            client.reject_capacity = False
+            pool.refresh_active_counts()
+            self.assertEqual(95, pool.old_model_submit_slots())
+            pool.add_task_for_model(
+                {"model": "eligible-old-model"},
+                model_last_modified="2025-01-01T00:00:00Z",
+                submitted_at="2026-10-02T00:00:00Z",
+            )
+            self.assertEqual(1, len(client.submitted))
+            restarted = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                capacity_state_path=state_path,
+            )
+            self.assertEqual([100], restarted.account_capacity_limits())
+
+    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)
@@ -291,8 +551,37 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
             recent_model_days=7,
         )
 
-        self.assertEqual(["configured_window", "last_7_days"], [stage["name"] for stage in stages])
+        self.assertEqual(
+            ["configured_window", "last_7_days", "recent_window_full"],
+            [stage["name"] for stage in stages],
+        )
         self.assertTrue(all(stage["updatedAfter"] >= now - timedelta(days=7) for stage in stages))
+        self.assertEqual(3000, stages[-1]["limit"])
+
+    def test_recent_only_scan_deepens_even_when_date_window_cannot_expand(self) -> None:
+        now = datetime(2026, 8, 11, tzinfo=timezone.utc)
+        for initial_cutoff in (None, now - timedelta(days=7), now - timedelta(days=30)):
+            with self.subTest(initial_cutoff=initial_cutoff):
+                stages = build_adaptive_scan_stages(
+                    now=now,
+                    initial_updated_after=initial_cutoff,
+                    initial_limit=20,
+                    max_models=800,
+                    allow_older_than_recent_window=False,
+                )
+                self.assertEqual([20, 800], [stage["limit"] for stage in stages])
+                self.assertTrue(all(stage["updatedAfter"] == now - timedelta(days=7) for stage in stages))
+
+    def test_recent_only_scan_deduplicates_an_already_full_window(self) -> None:
+        now = datetime(2026, 8, 11, tzinfo=timezone.utc)
+        stages = build_adaptive_scan_stages(
+            now=now,
+            initial_updated_after=now - timedelta(days=7),
+            initial_limit=100,
+            max_models=100,
+            allow_older_than_recent_window=False,
+        )
+        self.assertEqual(1, len(stages))
 
     def test_submit_candidate_reports_old_model_policy_skip_at_dynamic_threshold(self) -> None:
         client = FakeClient(active_count=95)
@@ -543,10 +832,59 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
             )
 
             self.assertEqual(2, summary["submittedCount"])
-            self.assertEqual(len(supported_gpus), summary["skipReasonCounts"]["already_processed_for_gpu"])
+            self.assertEqual(1, summary["skipReasonCounts"]["model_route_limit_reached"])
             self.assertEqual("all_history", summary["scanStages"][-1]["name"])
+            self.assertEqual("last_modified", summary["scanStages"][-1]["sort"])
             self.assertIsNone(discovery.calls[-1])
 
+    def test_reserved_slots_find_recent_candidates_beyond_first_scan_prefix(self) -> None:
+        class RecentOnlyClient(ExhaustedRecentClient):
+            def old_model_submit_slots(self) -> int:
+                return 0
+
+        class PagedRecentDiscovery(FakeDiscovery):
+            def __init__(self) -> None:
+                super().__init__(4)
+                self.models[0] = replace(self.models[0], repo_id="owner/recent-exhausted")
+                self.limits: list[int] = []
+
+            def list_recent_models(self, *, limit, **_kwargs):  # noqa: ANN003
+                self.limits.append(limit)
+                return self.models[:limit]
+
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            root = Path(temporary_dir)
+            args = build_parser().parse_args([
+                "--task-types", "text-generation", "--since-hours", "168",
+                "--limit", "1", "--max-scan-models", "4",
+                "--skip-outcome-sync", "--skip-history-archive",
+                "--disable-gpu-strategy",
+            ])
+            for field, filename in (
+                ("runs_dir", "runs"), ("ledger_path", "ledger.jsonl"),
+                ("outcomes_path", "outcomes.jsonl"), ("claims_path", "claims.jsonl"),
+                ("history_archive_path", "history.jsonl"),
+                ("submission_exclusions_path", "exclusions.jsonl"),
+            ):
+                setattr(args, field, str(root / filename))
+            selector = TemplateSelector()
+            discovery = PagedRecentDiscovery()
+            client = RecentOnlyClient(
+                available=2,
+                processed_gpus=set(selector.supported_target_gpus("text-generation", auto_only=True)),
+            )
+            summary = run_submission(
+                args,
+                now=datetime(2026, 1, 2, 12, tzinfo=timezone.utc),
+                hf_discovery=discovery,  # type: ignore[arg-type]
+                modelhub_client=client,  # type: ignore[arg-type]
+                template_selector=selector,
+            )
+            self.assertEqual([1, 4], discovery.limits)
+            self.assertEqual(2, summary["submittedCount"])
+            self.assertEqual("recent_window_full", summary["scanStages"][-1]["name"])
+            self.assertEqual(4, summary["scannedModels"])
+
     def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None:
         with tempfile.TemporaryDirectory() as temporary_dir:
             root = Path(temporary_dir)

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:
@@ -211,7 +219,7 @@ class MarketIntelligenceTests(unittest.TestCase):
             "frameworkStats": {
                 "text-generation": {
                     "gpu": {
-                        "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.10},
+                        "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.30},
                         "vllm-mlu": {"modelCount": 800, "wilsonLowerBound": 0.35},
                         "vllm-customized": {"modelCount": 2, "wilsonLowerBound": 0.90},
                     }
@@ -349,7 +357,7 @@ class MarketIntelligenceTests(unittest.TestCase):
             "frameworkStats": {
                 "text-generation": {
                     "Cambricon_mlu-370-x4": {
-                        "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.10},
+                        "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.30},
                         "future-engine": {
                             "modelCount": 1000,
                             "successCount": 500,
@@ -388,7 +396,7 @@ class MarketIntelligenceTests(unittest.TestCase):
             "frameworkStats": {
                 "text-generation": {
                     "Cambricon_mlu-370-x4": {
-                        "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.10},
+                        "vllm": {"modelCount": 1000, "wilsonLowerBound": 0.30},
                         "future-engine": {
                             "modelCount": 2,
                             "wilsonLowerBound": 0.90,
@@ -446,6 +454,124 @@ class MarketIntelligenceTests(unittest.TestCase):
         )
         self.assertIsNone(candidate)
 
+    def test_low_confidence_public_framework_is_rejected_even_with_many_samples(self) -> None:
+        manager = MarketIntelligenceManager("unused.json", framework_min_samples=100)
+        manager.state = {
+            "frameworkStats": {
+                "text-generation": {
+                    "Cambricon_mlu-370-x4": {
+                        "vllm": {"modelCount": 10_000, "wilsonLowerBound": 0.12},
+                    }
+                }
+            }
+        }
+        candidate = choose_candidate_for_gpu(
+            model=HFModelSummary(
+                repo_id="owner/model",
+                downloads=100,
+                last_modified=None,
+                pipeline_tag="text-generation",
+            ),
+            inspection=ModelInspection(repo_id="owner/model", weight_files=["model.safetensors"]),
+            template_selector=TemplateSelector(),
+            task_types=["text-generation"],
+            target_gpu="Cambricon_mlu-370-x4",
+            market_intelligence=manager,
+        )
+        self.assertIsNone(candidate)
+
+    def test_repeated_ambiguous_failures_open_operational_circuit(self) -> None:
+        manager = MarketIntelligenceManager("unused.json", framework_min_samples=100)
+        manager.state = {
+            "frameworkStats": {
+                "text-generation": {
+                    "Biren_166m": {
+                        "vllm_fix_tokenizer": {"modelCount": 10_000, "wilsonLowerBound": 0.35},
+                    }
+                }
+            }
+        }
+        manager.set_local_outcome_stats(
+            {
+                "recentCombinationStats": {
+                    "Biren_166m|vllm_fix_tokenizer|text-generation": {
+                        "successCount": 0,
+                        "failureCount": 5,
+                        "attributableFailureCount": 0,
+                        "consecutiveFailures": 0,
+                        "lastTerminalAt": datetime.now(timezone.utc).isoformat(),
+                    }
+                }
+            }
+        )
+        metadata = manager.framework_metadata(
+            "text-generation", "Biren_166m", "vllm_fix_tokenizer"
+        )
+        self.assertTrue(metadata["frameworkCircuitOpen"])
+        self.assertEqual(
+            "recent_terminal_success_below_10_percent",
+            metadata["frameworkCircuitReason"],
+        )
+
+    def test_low_sample_gpu_with_recent_success_and_proven_framework_remains_eligible(self) -> None:
+        manager = MarketIntelligenceManager("unused.json", framework_min_samples=100)
+        manager.state = {
+            "gpuStats": {
+                "Sunrise_pt-200-x1": {
+                    "canVerify": True,
+                    "submissionEligible": False,
+                    "recentTerminal": 10,
+                    "recentSuccess": 5,
+                    "recentWilsonLowerBound": 0.24,
+                    "healthFactor": 1.0,
+                    "selectionWeight": 0.5,
+                }
+            },
+            "frameworkStats": {
+                "text-generation": {
+                    "Sunrise_pt-200-x1": {
+                        "vllm_fix_tokenizer": {
+                            "modelCount": 6_000,
+                            "wilsonLowerBound": 0.41,
+                            "officialConfigValid": True,
+                        }
+                    }
+                }
+            },
+        }
+        self.assertEqual(
+            ["Sunrise_pt-200-x1"],
+            manager.eligible_gpus(["Sunrise_pt-200-x1"]),
+        )
+
+    def test_low_sample_gpu_does_not_bypass_absolute_quality_floor(self) -> None:
+        manager = MarketIntelligenceManager("unused.json", framework_min_samples=100)
+        manager.state = {
+            "gpuStats": {
+                "gpu-low-quality": {
+                    "canVerify": True,
+                    "submissionEligible": False,
+                    "recentTerminal": 10,
+                    "recentSuccess": 1,
+                    "recentWilsonLowerBound": 0.02,
+                    "healthFactor": 1.0,
+                    "selectionWeight": 0.5,
+                }
+            },
+            "frameworkStats": {
+                "text-generation": {
+                    "gpu-low-quality": {
+                        "vllm": {
+                            "modelCount": 6_000,
+                            "wilsonLowerBound": 0.41,
+                            "officialConfigValid": True,
+                        }
+                    }
+                }
+            },
+        }
+        self.assertEqual([], manager.eligible_gpus(["gpu-low-quality"]))
+
     def test_five_consecutive_local_failures_circuit_break_framework(self) -> None:
         manager = MarketIntelligenceManager("unused.json", framework_min_samples=100)
         manager.state = {

tests/test_modelscope_discovery.py

@@ -11,7 +11,7 @@ if str(PACKAGE_DIR) in sys.path:
     sys.path.remove(str(PACKAGE_DIR))
 sys.path.insert(0, str(PACKAGE_DIR))
 
-from hf_discovery import HuggingFaceDiscovery  # noqa: E402
+from hf_discovery import HuggingFaceDiscovery, inspect_repo_tree  # noqa: E402
 from http_json import HttpJsonError  # noqa: E402
 
 
@@ -58,6 +58,56 @@ class ModelMetadataClient:
 
 
 class ModelScopeDiscoveryTests(unittest.TestCase):
+    def test_listing_diagnostics_distinguish_filtered_prefix_from_exhaustion(self) -> None:
+        class FilteredPageClient:
+            def request_json(self, *_args, **_kwargs):
+                return {"data": {"models": [
+                    {"id": f"owner/model-{index}", "downloads": 0}
+                    for index in range(50)
+                ]}}
+
+        discovery = HuggingFaceDiscovery(http_client=FilteredPageClient())  # type: ignore[arg-type]
+        self.assertEqual([], discovery.list_recent_models(
+            pipeline_tags=["text-generation"], limit=50, min_downloads=50,
+        ))
+        diagnostic = discovery.last_listing_diagnostics[0]
+        self.assertEqual(50, diagnostic["rawRowsScanned"])
+        self.assertEqual(0, diagnostic["acceptedModels"])
+        self.assertEqual("page_budget", diagnostic["stopReason"])
+        self.assertFalse(diagnostic["catalogExhausted"])
+
+    def test_failed_repo_tree_is_retried_instead_of_cached_as_empty(self) -> None:
+        class RecoveringClient:
+            calls = 0
+
+            def request_json(self, *_args, **_kwargs):
+                self.calls += 1
+                if self.calls == 1:
+                    raise HttpJsonError("temporary outage", status_code=503)
+                return {"Data": {"Files": [{"Path": "model.safetensors", "Size": 1024}]}}
+
+        client = RecoveringClient()
+        discovery = HuggingFaceDiscovery(legacy_http_client=client)  # type: ignore[arg-type]
+        with self.assertRaises(HttpJsonError):
+            discovery.list_repo_tree("owner/recoverable")
+        self.assertNotIn("owner/recoverable", discovery._repo_tree_cache)
+        entries = discovery.list_repo_tree("owner/recoverable")
+        self.assertEqual("model.safetensors", entries[0]["Path"])
+        self.assertEqual(entries, discovery.list_repo_tree("owner/recoverable"))
+        self.assertEqual(2, client.calls)
+
+    def test_duplicate_normalized_paths_do_not_inflate_repository_size(self) -> None:
+        gib = 1024 ** 3
+        inspection = inspect_repo_tree("owner/model", [
+            {"Path": "model.safetensors", "Size": 20 * gib},
+            {"Path": "./model.safetensors", "Size": 19 * gib},
+            {"Path": "/model.safetensors", "Size": 20 * gib},
+        ])
+        self.assertEqual(["model.safetensors"], inspection.file_paths)
+        self.assertEqual(["model.safetensors"], inspection.weight_files)
+        self.assertEqual(20 * gib, inspection.repository_size_bytes)
+        self.assertEqual(20 * gib, inspection.estimated_load_bytes("vllm"))
+
     def test_model_last_modified_uses_detail_api_and_is_cached(self) -> None:
         metadata_client = ModelMetadataClient()
         discovery = HuggingFaceDiscovery(legacy_http_client=metadata_client)  # type: ignore[arg-type]
@@ -93,6 +143,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,
@@ -20,6 +21,7 @@ from modelhub_client import (  # noqa: E402
     parse_model_submission_precheck,
 )
 from models import HFModelSummary, ModelInspection  # noqa: E402
+from outcome_tracker import OutcomeTracker  # noqa: E402
 from submission_exclusions import SubmissionExclusionStore  # noqa: E402
 from template_selector import TemplateSelector  # noqa: E402
 
@@ -68,6 +70,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 +119,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 +136,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 +262,202 @@ 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(
+            {"failed", "running", "success", "unknown", "failed_without_framework"},
+            precheck["processedGpus"],
+        )
+        self.assertEqual({"success"}, precheck["successfulGpus"])
+        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_remains_closed_even_when_an_alternative_framework_exists(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("prior_failed_gpu_closed_accuracy_first", skipped[-1]["reason"])
+        self.assertEqual([], failed)
+        self.assertEqual([], candidates)
+
+        blocked = submit_candidate(candidate("Cambricon_mlu-370-x4"), client)  # type: ignore[arg-type]
+        self.assertEqual("duplicate", blocked["outcome"])
+        self.assertEqual("prior_failed_gpu_closed_accuracy_first", blocked["reason"])
+        self.assertEqual(0, 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_gpu_closed_accuracy_first", 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_closed_accuracy_first",
+            no_market_skips[-1]["reason"],
+        )
+
+    def test_model_route_limit_is_global_across_poll_cycles(self) -> None:
+        class SixRouteClient:
+            @staticmethod
+            def model_submission_precheck(_model_id: str) -> dict:
+                return {
+                    "processedGpus": {f"gpu-{index}" for index in range(1, 7)},
+                    "successfulGpus": {"gpu-1"},
+                    "failedFrameworksByGpu": {},
+                }
+
+        candidates, skipped, failed = process_model_for_candidates(
+            model=SafeDiscovery().model,
+            hf_discovery=SafeDiscovery(),  # type: ignore[arg-type]
+            modelhub_client=SixRouteClient(),  # type: ignore[arg-type]
+            template_selector=TemplateSelector(),
+            target_gpus=["gpu-7"],
+            allowed_task_types=["text-generation"],
+        )
+        self.assertEqual([], candidates)
+        self.assertEqual([], failed)
+        self.assertEqual("model_route_limit_reached", skipped[0]["reason"])
+        self.assertEqual(6, skipped[0]["processedGpuCount"])
+
+    def test_model_waits_for_feedback_after_two_pending_routes(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            root = Path(temporary_dir)
+            tracker = OutcomeTracker(
+                root / "outcomes.jsonl",
+                checkpoint_path=root / "checkpoint.json",
+                recent_path=root / "recent.jsonl",
+            )
+            tracker.record_submission(
+                "owner/model", "gpu-1", "vllm", "text-generation", "task-1",
+                "2026-09-22T00:00:00+00:00",
+            )
+            tracker.record_submission(
+                "owner/model", "gpu-2", "vllm", "text-generation", "task-2",
+                "2026-09-22T00:01:00+00:00",
+            )
+
+            candidates, skipped, failed = process_model_for_candidates(
+                model=SafeDiscovery().model,
+                hf_discovery=SafeDiscovery(),  # type: ignore[arg-type]
+                modelhub_client=OtherGpuProcessedClient(),  # type: ignore[arg-type]
+                template_selector=TemplateSelector(),
+                target_gpus=["gpu-3"],
+                allowed_task_types=["text-generation"],
+                outcome_tracker=tracker,
+            )
+
+            self.assertEqual([], candidates)
+            self.assertEqual([], failed)
+            self.assertEqual("model_pending_feedback_limit", skipped[0]["reason"])
+            self.assertEqual(2, skipped[0]["pendingRouteCount"])
+
+    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):