Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
964362818e | ||
|
|
fb3c1ab158 | ||
|
|
dc8365c215 | ||
|
|
6154dc738f | ||
|
|
4afce879dc |
82
README.md
82
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
|
||||
@@ -464,11 +513,28 @@ 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-v35 -m "ModelHub agent 2026.09.20.1"
|
||||
git push origin main agent-v35
|
||||
git tag -a agent-v36 -m "ModelHub agent 2026.09.20.2"
|
||||
git push origin main agent-v36
|
||||
```
|
||||
|
||||
@@ -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,28 +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 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.
|
||||
- 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
|
||||
@@ -144,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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -517,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
|
||||
@@ -569,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 (
|
||||
@@ -579,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]] = []
|
||||
@@ -602,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
|
||||
@@ -639,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
|
||||
|
||||
@@ -705,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,
|
||||
@@ -716,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 = (
|
||||
@@ -726,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)
|
||||
@@ -807,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:
|
||||
@@ -948,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 = {
|
||||
@@ -1050,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}",
|
||||
@@ -1150,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}
|
||||
@@ -1499,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'}",
|
||||
@@ -1601,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(
|
||||
@@ -1743,7 +1805,7 @@ def run_submission(
|
||||
}
|
||||
)
|
||||
continue
|
||||
if result["outcome"] == "framework_evidence_deferred":
|
||||
if result["outcome"] in {"framework_evidence_deferred", "rate_limit_deferred"}:
|
||||
batch_policy_skipped_candidates.append(candidate)
|
||||
skipped.append({
|
||||
"repoId": candidate["repoId"],
|
||||
@@ -1906,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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -37,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)
|
||||
@@ -47,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:
|
||||
@@ -62,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,
|
||||
@@ -310,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,
|
||||
@@ -409,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()
|
||||
@@ -447,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,
|
||||
@@ -472,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
|
||||
@@ -521,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:
|
||||
@@ -534,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)
|
||||
},
|
||||
}
|
||||
@@ -552,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)
|
||||
@@ -576,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
|
||||
@@ -606,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])
|
||||
@@ -694,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))
|
||||
@@ -781,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 (
|
||||
@@ -797,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(
|
||||
@@ -819,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
|
||||
@@ -844,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] = {
|
||||
@@ -872,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()
|
||||
@@ -890,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),
|
||||
@@ -905,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)
|
||||
@@ -943,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,
|
||||
@@ -977,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:
|
||||
@@ -986,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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)),
|
||||
@@ -862,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 (
|
||||
@@ -1035,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'}"
|
||||
)
|
||||
|
||||
@@ -1079,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']} "
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -1 +1 @@
|
||||
AGENT_VERSION = "2026.09.20.1"
|
||||
AGENT_VERSION = "2026.09.22.1"
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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,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)
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user