4 Commits

Author SHA1 Message Date
coolboy
964362818e 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.
2026-10-03 01:37:55 +08:00
coolboy
fb3c1ab158 feat: pack recent submissions into reserved account slots first
Preserve shared queue positions for old-model candidates with constrained
admission. Retain headroom balancing for ties and weekly calibration pinning.
Add offline mixed-age allocation regressions.
2026-10-02 21:43:50 +08:00
coolboy
dc8365c215 fix: prevent stale capacity poisoning and deepen recent-only refill scans
Keep ordinary capacity races from lowering persistent per-account limits.
Deepen recent-only discovery within the existing age and quality gates.
Add offline regressions for recovery and reserved-slot refill.
2026-10-02 21:24:08 +08:00
CoolBoy
6154dc738f fix: restore accuracy-first route admission 2026-09-22 15:26:42 +08:00
16 changed files with 709 additions and 107 deletions

View File

@@ -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
@@ -296,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
@@ -481,6 +518,18 @@ 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 "我的适配智能体".

View File

@@ -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,20 +118,22 @@ 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 the first 3,000 raw ModelScope rows ordered by update time. If the
candidate pool is still short, it rotates through up to 200 download-ranked rows per cycle
as an older-model fallback.
- 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. `未验证` with no records is eligible. `验证失败` is eligible only with a
different publicly vetted framework after all recorded attempts explicitly failed;
`已验证`, `验证中`, and ambiguous records remain blocked. Every actual submission
performs a fresh uncached check for its exact GPU and framework.
- 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.
- The same framework is not retried for a model/GPU after a recorded failure.
- 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.

View File

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

View File

@@ -110,6 +110,7 @@ class HuggingFaceDiscovery:
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:
@@ -133,6 +134,7 @@ class HuggingFaceDiscovery:
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))
@@ -156,6 +158,7 @@ class HuggingFaceDiscovery:
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")) > (
@@ -175,6 +178,7 @@ class HuggingFaceDiscovery:
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)
@@ -183,12 +187,23 @@ class HuggingFaceDiscovery:
filter_disabled = self._unsupported_task_filters.get(task_tag, 0.0) > time.monotonic()
models: list[HFModelSummary] = []
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:
@@ -211,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",
@@ -223,12 +239,15 @@ class HuggingFaceDiscovery:
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)
@@ -238,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(
@@ -249,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
@@ -412,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)
@@ -570,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()
@@ -584,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)),
)

View File

@@ -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:
@@ -376,6 +377,7 @@ def choose_candidate_for_gpu_detailed(
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:
@@ -472,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)
@@ -572,9 +591,11 @@ 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 (
@@ -582,6 +603,31 @@ def process_model_for_candidates(
[{"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]] = []
@@ -605,33 +651,14 @@ def process_model_for_candidates(
if target_gpu in processed_gpus:
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"})
continue
if target_gpu in failed_frameworks_by_gpu and market_intelligence is None:
if target_gpu in failed_frameworks_by_gpu:
skipped.append({
"repoId": model.repo_id,
"targetGpu": target_gpu,
"reason": "prior_failed_gpu_requires_public_framework_evidence",
"reason": "prior_failed_gpu_closed_accuracy_first",
})
continue
failed_frameworks = set(failed_frameworks_by_gpu.get(target_gpu) or set())
if (
failed_frameworks
and market_intelligence is not None
and hasattr(market_intelligence, "has_vetted_alternative_framework")
and allowed_task_types_set
and all(market_intelligence.has_framework_route(task_type, target_gpu) for task_type in allowed_task_types_set)
and not any(
market_intelligence.has_vetted_alternative_framework(
task_type, target_gpu, failed_frameworks
)
for task_type in allowed_task_types_set
)
):
skipped.append({
"repoId": model.repo_id,
"targetGpu": target_gpu,
"reason": "prior_failed_framework_no_alternative",
})
continue
if outcome_tracker and outcome_tracker.is_model_gpu_failed(model.repo_id, target_gpu):
skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"})
continue
@@ -642,6 +669,14 @@ def process_model_for_candidates(
"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
@@ -708,6 +743,7 @@ def process_model_for_candidates(
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,
@@ -719,6 +755,7 @@ def process_model_for_candidates(
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 = (
@@ -729,11 +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)
@@ -810,20 +853,17 @@ def build_adaptive_scan_stages(
"limit": max_models,
}
)
if allow_older_than_recent_window:
# ModelScope only exposes the first 3,000 raw rows for a sort order.
# Downloads ordering reaches a different older, established slice
# after the recent-update slice has produced too few candidates.
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": "popular_history",
"updatedAfter": None,
"limit": min(max_models, 200),
"sort": "downloads",
"rotatePages": True,
"name": "recent_window_full",
"updatedAfter": recent_cutoff,
"limit": max_models,
}
)
deduped: list[dict[str, Any]] = []
seen: set[tuple[str | None, int, str]] = set()
for stage in stages:
@@ -951,20 +991,20 @@ def submit_candidate(
"candidate": candidate,
"reason": "already_processed_for_gpu",
}
failed_frameworks = set(
(precheck.get("failedFrameworksByGpu") or {}).get(candidate["targetGpu"]) or set()
)
if candidate["framework"] in failed_frameworks:
if len(set(precheck.get("processedGpus") or set())) >= MAX_ROUTES_PER_MODEL:
return {
"outcome": "duplicate",
"candidate": candidate,
"reason": "prior_failed_framework_for_gpu",
"reason": "model_route_limit_reached",
}
if failed_frameworks and not candidate.get("frameworkEvidenceQualified"):
failed_frameworks = set(
(precheck.get("failedFrameworksByGpu") or {}).get(candidate["targetGpu"]) or set()
)
if failed_frameworks:
return {
"outcome": "framework_evidence_deferred",
"outcome": "duplicate",
"candidate": candidate,
"reason": "prior_failed_gpu_requires_public_framework_evidence",
"reason": "prior_failed_gpu_closed_accuracy_first",
}
payload = {
@@ -1622,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(

View File

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

View File

@@ -38,6 +38,7 @@ def parse_model_submission_precheck(payload: Any) -> dict[str, Any]:
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)
@@ -48,6 +49,12 @@ def parse_model_submission_precheck(payload: Any) -> dict[str, Any]:
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:
@@ -63,20 +70,26 @@ def parse_model_submission_precheck(payload: Any) -> dict[str, Any]:
frameworks.add(framework)
else:
failed_frameworks_by_gpu[name] = frameworks
continue
# Verified, validating and unknown states remain closed. A failed GPU
# opens only for a different framework with explicit failed records.
blocked_gpus.add(name)
# 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": 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,
@@ -935,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(
@@ -1016,7 +1049,12 @@ 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

View File

@@ -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

View File

@@ -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

View File

@@ -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),

View File

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

View File

@@ -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

View File

@@ -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
@@ -216,6 +217,75 @@ 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] = []
@@ -481,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)
@@ -733,11 +832,59 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
)
self.assertEqual(2, summary["submittedCount"])
self.assertEqual(len(supported_gpus), summary["skipReasonCounts"]["already_processed_for_gpu"])
self.assertEqual("popular_history", summary["scanStages"][-1]["name"])
self.assertEqual("downloads", summary["scanStages"][-1]["sort"])
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)

View File

@@ -219,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},
}
@@ -357,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,
@@ -396,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,
@@ -454,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 = {

View File

@@ -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]

View File

@@ -21,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
@@ -286,9 +287,10 @@ class SubmissionSafetyTests(unittest.TestCase):
}
precheck = parse_model_submission_precheck(payload)
self.assertEqual(
{"running", "success", "unknown", "failed_without_framework"},
{"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:
@@ -321,7 +323,7 @@ class SubmissionSafetyTests(unittest.TestCase):
self.assertEqual([], active_candidates)
self.assertEqual("already_processed_for_gpu", active_skips[0]["reason"])
def test_failed_gpu_requires_an_alternative_framework(self) -> None:
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(
@@ -334,23 +336,14 @@ class SubmissionSafetyTests(unittest.TestCase):
market_intelligence=TwoFrameworkMarket(), # type: ignore[arg-type]
official_registry=AllGpuOfficialRegistry(), # type: ignore[arg-type]
)
self.assertEqual([], skipped)
self.assertEqual("prior_failed_gpu_closed_accuracy_first", skipped[-1]["reason"])
self.assertEqual([], failed)
self.assertEqual(["vllm-mlu"], [item["framework"] for item in candidates])
self.assertEqual(["vllm"], candidates[0]["previousFailedFrameworks"])
self.assertEqual([], candidates)
blocked = submit_candidate({**candidates[0], "framework": "vllm"}, client) # type: ignore[arg-type]
accepted = submit_candidate(candidates[0], client) # type: ignore[arg-type]
blocked = submit_candidate(candidate("Cambricon_mlu-370-x4"), client) # type: ignore[arg-type]
self.assertEqual("duplicate", blocked["outcome"])
self.assertEqual("prior_failed_framework_for_gpu", blocked["reason"])
self.assertEqual("submitted", accepted["outcome"])
self.assertEqual(1, client.add_calls)
unvetted = submit_candidate(
{**candidates[0], "frameworkEvidenceQualified": False}, client # type: ignore[arg-type]
)
self.assertEqual("framework_evidence_deferred", unvetted["outcome"])
self.assertEqual(1, client.add_calls)
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,
@@ -363,7 +356,7 @@ class SubmissionSafetyTests(unittest.TestCase):
official_registry=AllGpuOfficialRegistry(), # type: ignore[arg-type]
)
self.assertEqual([], no_routes)
self.assertEqual("prior_failed_framework_no_alternative", no_route_skips[-1]["reason"])
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,
@@ -375,10 +368,65 @@ class SubmissionSafetyTests(unittest.TestCase):
)
self.assertEqual([], no_market_routes)
self.assertEqual(
"prior_failed_gpu_requires_public_framework_evidence",
"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