3 Commits

Author SHA1 Message Date
coolboy
dc8365c215 fix: prevent stale capacity poisoning and deepen recent-only refill scans
Keep ordinary capacity races from lowering persistent per-account limits.
Deepen recent-only discovery within the existing age and quality gates.
Add offline regressions for recovery and reserved-slot refill.
2026-10-02 21:24:08 +08:00
CoolBoy
6154dc738f fix: restore accuracy-first route admission 2026-09-22 15:26:42 +08:00
CoolBoy
4afce879dc feat: calibrate account capacity weekly 2026-09-20 13:03:35 +08:00
12 changed files with 942 additions and 132 deletions

View File

@@ -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,6 +301,16 @@ 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.
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`
@@ -464,11 +491,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
```

View File

@@ -79,21 +79,25 @@ bash run_poll.sh --dry-run
## Behavior
- The runner auto-discovers all safe GPU/template combinations from the public submit catalog.
- Automatic GPU selection uses exact 70/30 accepted-task scheduling: long-term
Wilson-ranked top 3 GPUs and the top GPUs from the latest 1,000 terminal tasks.
There is no all-GPU exploration category.
- Within each category, weighted-fair scheduling uses estimated queue backlog hours,
- Automatic GPU selection uses one success-first score built from qualified
local cohorts and ModelHub public evidence. There is no random or fixed-quota
exploration category.
- Within the qualified success band, scheduling uses estimated queue backlog hours,
recent public throughput/success, machine availability, and worker concurrency.
Unavailable or stalled GPU pools are circuit-broken instead of continuing to absorb work.
- Compatible frameworks are ranked by ModelHub public aggregate success statistics
plus capped local GPU+framework evidence, with a 300-sample public minimum and
Wilson confidence bounds. Missing or undersized public evidence receives zero
traffic rather than falling back to exploration.
a 20% Wilson lower-bound floor. GPU pools normally require a 15% floor.
Missing or undersized public evidence receives zero traffic rather than
falling back to exploration; a low-sample GPU requires a recent success and a
framework that already passes the stricter framework gate.
- New frameworks are discovered from the live catalog but get no novelty bonus.
They are eligible only with a complete official build config that passes local
validation and a confidence score at least 10% above the best incumbent.
- Five consecutive local failures pause a GPU/framework pair for 12 hours; a
sub-20% rate over the latest 20 terminal tasks pauses it for 6 hours.
- Five consecutive local failures pause a GPU/framework pair for 12 hours. Five
or more recent terminal outcomes below 10% success also pause it for 12 hours,
including ambiguous runtime failures. A sub-20% attributable rate over the
latest 20 terminal tasks pauses it for 6 hours.
- An explicit "framework does not support this model/architecture" failure learns
a 30-day GPU + framework + task + architecture block. Architecture identity
comes from candidate `config.json` plus exact unsupported `model_type` or
@@ -114,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.

View File

@@ -66,6 +66,7 @@ DEFAULT_LEDGER_PATH = Path("ledger/submissions.jsonl")
ADAPTIVE_SCAN_MAX_MODELS = 3000
ADAPTIVE_SCAN_MIN_FALLBACK_MODELS = 500
MAX_ROUTES_PER_MODEL = 6
MAX_PENDING_ROUTES_PER_MODEL = 2
def build_parser() -> argparse.ArgumentParser:
@@ -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)
@@ -517,6 +518,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 +573,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 +585,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 +633,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
@@ -731,6 +743,9 @@ def process_model_for_candidates(
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 +822,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 +960,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 +1062,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 +1173,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 +1523,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'}",
@@ -1743,7 +1773,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 +1936,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,

View File

@@ -20,7 +20,8 @@ DEFAULT_THROUGHPUT_WINDOW_HOURS = 6
DEFAULT_FETCH_WORKERS = 4
DEFAULT_FRAMEWORK_MIN_SAMPLES = 300
DEFAULT_GPU_MIN_RECENT_TERMINALS = 20
DEFAULT_FRAMEWORK_MIN_WILSON = 0.05
DEFAULT_FRAMEWORK_MIN_WILSON = 0.20
DEFAULT_GPU_MIN_WILSON = 0.15
DEFAULT_COMMUNITY_WINDOW_DAYS = 30
DEFAULT_COMMUNITY_MAX_MODELS = 200
DEFAULT_COMMUNITY_HYDRATE_PER_REFRESH = 50
@@ -201,7 +202,7 @@ class MarketIntelligenceManager:
and health_factor >= 0.5
and int(item.get("recentTerminal") or 0) >= DEFAULT_GPU_MIN_RECENT_TERMINALS
and int(item.get("recentSuccess") or 0) > 0
and wilson >= DEFAULT_FRAMEWORK_MIN_WILSON
and wilson >= DEFAULT_GPU_MIN_WILSON
)
def _base_state(self, gpus: list[str], task_types: list[str], now: datetime) -> dict[str, Any]:
@@ -554,7 +555,7 @@ class MarketIntelligenceManager:
and health_factor >= 0.5
and int(stats.get("recentTerminal") or 0) >= DEFAULT_GPU_MIN_RECENT_TERMINALS
and int(stats.get("recentSuccess") or 0) > 0
and wilson >= DEFAULT_FRAMEWORK_MIN_WILSON
and wilson >= DEFAULT_GPU_MIN_WILSON
)
result[gpu] = stats
@@ -661,6 +662,25 @@ 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:
low_sample = int(item.get("recentTerminal") or 0) < DEFAULT_GPU_MIN_RECENT_TERMINALS
has_recent_success = int(item.get("recentSuccess") or 0) > 0
recent_wilson = float(item.get("recentWilsonLowerBound") or 0.0)
healthy = float(item.get("healthFactor") or 1.0) >= 0.5
has_proven_framework = any(
bool(framework.get("officialConfigValid", False))
and int(framework.get("modelCount") or 0) >= self.framework_min_samples
and float(framework.get("wilsonLowerBound") or 0.0) >= DEFAULT_FRAMEWORK_MIN_WILSON
for by_gpu in ((self.state or {}).get("frameworkStats") or {}).values()
for framework in ((by_gpu or {}).get(gpu) or {}).values()
)
if (
item.get("canVerify") is False
or not low_sample
or not has_recent_success
or recent_wilson < DEFAULT_GPU_MIN_WILSON
or not healthy
or not has_proven_framework
):
continue
ranked.append((-float(item.get("selectionWeight") or 1.0), index, gpu))
return [gpu for _weight, _index, gpu in sorted(ranked)]
@@ -723,6 +743,7 @@ class MarketIntelligenceManager:
recent_item = (self.local_outcome_stats.get("recentCombinationStats") or {}).get(local_key) or {}
recent_success = max(0, int(recent_item.get("successCount") or 0))
recent_all_failures = max(0, int(recent_item.get("failureCount") or 0))
recent_failure = max(
0,
int(recent_item.get("attributableFailureCount", recent_item.get("failureCount") or 0)),
@@ -743,6 +764,16 @@ class MarketIntelligenceManager:
elif last_terminal_at is not None and consecutive_failures >= 5:
circuit_reason = "five_consecutive_local_failures"
circuit_until = last_terminal_at + timedelta(hours=12)
elif (
last_terminal_at is not None
and recent_success + recent_all_failures >= 5
and recent_success / (recent_success + recent_all_failures) < 0.10
):
# Runtime-empty and other unresolved failures must not create a
# permanent compatibility blacklist. Repetition is still strong
# operational evidence, so pause the route while feedback settles.
circuit_reason = "recent_terminal_success_below_10_percent"
circuit_until = last_terminal_at + timedelta(hours=12)
elif last_terminal_at is not None and recent_samples >= 20 and recent_rate is not None and recent_rate < 0.20:
circuit_reason = "recent_local_success_below_20_percent"
circuit_until = last_terminal_at + timedelta(hours=6)

View File

@@ -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 (
@@ -819,14 +970,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,6 +1000,7 @@ class ModelHubClientPool:
key=lambda index: (index - self._selection_cursor) % len(self.clients),
)
self._selection_cursor = (selected_index + 1) % len(self.clients)
if weekly_target is None:
self._capacity_probe_attempted.add(selected_index)
self._reservation_sequence += 1
reservation_id = self._reservation_sequence
@@ -872,12 +1029,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 +1057,16 @@ class ModelHubClientPool:
if discovered_cap <= self._account_caps[index]:
return
self._account_caps[index] = discovered_cap
weekly = self._weekly_target_locked() == index
self._persist_account_caps()
if not weekly or discovered_cap % 25 == 0:
print(f"[capacity] account={index + 1:02d} discovered_cap={discovered_cap}", flush=True)
def old_model_submit_slots(self) -> int:
"""Return queue positions where models older than the recent window are allowed."""
self._refresh_active_counts()
with self._state_lock:
return sum(
normal_slots = sum(
max(
0,
self._old_model_count_limit_locked(index) - self._effective_count_locked(index),
@@ -905,6 +1074,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 +1113,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 +1153,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 +1165,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

View File

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

View File

@@ -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']} "

View File

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

View File

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

View File

@@ -6,6 +6,7 @@ import tempfile
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
@@ -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,227 @@ def make_candidate(index: int) -> dict:
class ClientPoolConcurrencyTests(unittest.TestCase):
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 +514,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 +795,59 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
)
self.assertEqual(2, summary["submittedCount"])
self.assertEqual(len(supported_gpus), summary["skipReasonCounts"]["already_processed_for_gpu"])
self.assertEqual("popular_history", summary["scanStages"][-1]["name"])
self.assertEqual("downloads", summary["scanStages"][-1]["sort"])
self.assertEqual(1, summary["skipReasonCounts"]["model_route_limit_reached"])
self.assertEqual("all_history", summary["scanStages"][-1]["name"])
self.assertEqual("last_modified", summary["scanStages"][-1]["sort"])
self.assertIsNone(discovery.calls[-1])
def test_reserved_slots_find_recent_candidates_beyond_first_scan_prefix(self) -> None:
class RecentOnlyClient(ExhaustedRecentClient):
def old_model_submit_slots(self) -> int:
return 0
class PagedRecentDiscovery(FakeDiscovery):
def __init__(self) -> None:
super().__init__(4)
self.models[0] = replace(self.models[0], repo_id="owner/recent-exhausted")
self.limits: list[int] = []
def list_recent_models(self, *, limit, **_kwargs): # noqa: ANN003
self.limits.append(limit)
return self.models[:limit]
with tempfile.TemporaryDirectory() as temporary_dir:
root = Path(temporary_dir)
args = build_parser().parse_args([
"--task-types", "text-generation", "--since-hours", "168",
"--limit", "1", "--max-scan-models", "4",
"--skip-outcome-sync", "--skip-history-archive",
"--disable-gpu-strategy",
])
for field, filename in (
("runs_dir", "runs"), ("ledger_path", "ledger.jsonl"),
("outcomes_path", "outcomes.jsonl"), ("claims_path", "claims.jsonl"),
("history_archive_path", "history.jsonl"),
("submission_exclusions_path", "exclusions.jsonl"),
):
setattr(args, field, str(root / filename))
selector = TemplateSelector()
discovery = PagedRecentDiscovery()
client = RecentOnlyClient(
available=2,
processed_gpus=set(selector.supported_target_gpus("text-generation", auto_only=True)),
)
summary = run_submission(
args,
now=datetime(2026, 1, 2, 12, tzinfo=timezone.utc),
hf_discovery=discovery, # type: ignore[arg-type]
modelhub_client=client, # type: ignore[arg-type]
template_selector=selector,
)
self.assertEqual([1, 4], discovery.limits)
self.assertEqual(2, summary["submittedCount"])
self.assertEqual("recent_window_full", summary["scanStages"][-1]["name"])
self.assertEqual(4, summary["scannedModels"])
def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None:
with tempfile.TemporaryDirectory() as temporary_dir:
root = Path(temporary_dir)

View File

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

View File

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