diff --git a/README.md b/README.md index 41a22dc437..1d05e72b01 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ Optional tuning: - The hosted entrypoint always uses `--max-submits-per-run 0` so stale deployment settings cannot restrict a refill cycle to five submissions. - `MODELHUB_AGENT_ACTIVE_TASK_CAP` default `100` per account -- `MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES` default `3` +- `MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES` is a legacy setting ignored by the hosted worker; it uses weekly calibration only - `MODELHUB_CAPACITY_STATE_PATH` default `.modelhub_state/account_capacity.json` - `MODELHUB_AGENT_ACTIVE_COUNTS_TTL_SECONDS` default `15` - `MODELHUB_AGENT_RESERVATION_TTL_SECONDS` default `120` @@ -228,13 +228,25 @@ or 500, older models stop at positions 95, 195, or 495 respectively. Old-model reservations are made under the same account lock as capacity reservations, so concurrent submissions cannot enter the reserved suffix. Once every account's old-model allowance is exhausted, discovery is capped at the seven-day window. -Unknown model timestamps are treated as old for new submissions. When capacity -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 @@ -464,11 +476,16 @@ Architecture history backfill now sends the date format accepted by ModelHub. The full-repository-size OOM rule is unchanged because ModelHub's own preflight uses that size. +Version `2026.09.20.2` replaces routine capacity probes with per-account +calibration each Monday 10:00-10:59 Asia/Shanghai. During that window only, +the worker may borrow the five recent-model positions to measure a new limit; +ordinary capacity-minus-five admission resumes afterward. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-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 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 55ea2eb683..d3d3050883 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -134,8 +134,11 @@ bash run_poll.sh --dry-run - 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 +147,7 @@ bash run_poll.sh --dry-run Incomplete size/capacity evidence is never used for cancellation. - Models older than seven days may occupy only the current account capacity minus its final 5 positions. The account pool enforces this per-account admission boundary atomically and updates - it when capacity probing discovers a higher limit. A failed active-count read makes that account + it when weekly calibration discovers a higher limit. A failed active-count read makes that account ineligible for older models while other readable accounts are still tried. Recent models may use all available positions. Age policy never cancels an existing waiting or running task; startup and scheduled cleanup remain limited to deterministic OOM and learned architecture evidence. diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 89386ce858..120ac2ac86 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -258,7 +258,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--capacity-probe-interval-cycles", type=int, - default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "3")), + default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "0")), help=argparse.SUPPRESS, ) parser.add_argument("--hf-base-url", default=os.getenv("MODELSCOPE_BASE_URL", "https://modelscope.cn"), help=argparse.SUPPRESS) @@ -517,6 +517,9 @@ def resolve_submit_concurrency( modelhub_client: ModelHubClient | ModelHubClientPool | Any, planned_submit_count: int, ) -> int: + if hasattr(modelhub_client, "weekly_calibration_active") and modelhub_client.weekly_calibration_active(): + # Pinning and capacity evidence require one in-flight task per account. + return 1 explicit_workers = int(getattr(args, "submit_concurrency", 0) or 0) if explicit_workers > 0: return explicit_workers @@ -1050,6 +1053,17 @@ def submit_candidate( "candidate": candidate, "reason": "account_capacity_saturated", } + if exc.code == 429: + print( + f"[submit] deferred repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=platform_rate_limited", + flush=True, + ) + return { + "outcome": "rate_limit_deferred", + "candidate": candidate, + "reason": "platform_rate_limited", + } print( f"[submit] failed repo={candidate['repoId']} gpu={candidate['targetGpu']} " f"framework={candidate['framework']} reason={exc}", @@ -1150,11 +1164,12 @@ def run_submission( clients = [ModelHubClient(token=token, base_url=args.modelhub_base_url) for token in token_values] modelhub_client = ModelHubClientPool( clients, - capacity_probe_interval_cycles=max(0, int(getattr(args, "capacity_probe_interval_cycles", 3) or 0)), + capacity_probe_interval_cycles=0, capacity_state_path=Path(getattr(args, "capacity_state_path", DEFAULT_CAPACITY_STATE_PATH)), recent_model_reserve_slots=max(0, int(getattr(args, "recent_model_reserve_slots", 5) or 0)), recent_model_days=max(1, int(getattr(args, "recent_model_days", 7) or 7)), ) + modelhub_client.configure_weekly_calibration(now) official_registry: OfficialCapabilityRegistry | None = None official_summary: dict[str, Any] = {"enabled": False} @@ -1499,10 +1514,16 @@ def run_submission( old_model_queue_thresholds: list[int | None] | None = None if hasattr(modelhub_client, "old_model_queue_thresholds"): old_model_queue_thresholds = list(modelhub_client.old_model_queue_thresholds()) + weekly_calibration = ( + modelhub_client.weekly_calibration_status() + if hasattr(modelhub_client, "weekly_calibration_status") else {} + ) allow_older_models_for_scan = old_model_slots_before_scan is None or old_model_slots_before_scan > 0 print( - f"[age-policy] mode=admission_only recent_days={recent_model_days} " + f"[age-policy] mode={'weekly_capacity_calibration' if weekly_calibration.get('targetAccount') else 'admission_only'} " + f"recent_days={recent_model_days} " f"reserve_recent_slots={recent_model_reserve_slots} " + f"calibration_account={weekly_calibration.get('targetAccount') or 'n/a'} " f"account_thresholds={','.join('n/a' if value is None else str(value) for value in old_model_queue_thresholds) if old_model_queue_thresholds is not None else 'n/a'} " f"old_model_slots={old_model_slots_before_scan if old_model_slots_before_scan is not None else 'n/a'} " f"scan_older={'on' if allow_older_models_for_scan else 'off'}", @@ -1743,7 +1764,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 +1927,15 @@ def run_submission( "marketIntelligence": market_summary, "candidatePreflight": preflight_summary, "agePolicy": { - "mode": "admission_only", + "mode": "weekly_capacity_calibration" if weekly_calibration.get("targetAccount") else "admission_only", "recentModelDays": recent_model_days, "recentModelReserveSlots": recent_model_reserve_slots, "oldModelSubmitThresholds": old_model_queue_thresholds, "oldModelSlotsBeforeScan": old_model_slots_before_scan, "olderHistoryScanEnabled": allow_older_models_for_scan, + "calibrationAccount": weekly_calibration.get("targetAccount"), }, + "weeklyCalibration": weekly_calibration, "scanLimit": scan_limit, "rawCandidateGoal": raw_candidate_goal, "candidatePoolLimit": candidate_pool_limit, diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 2e43a76ed6..4ff3273db5 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -8,6 +8,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta from pathlib import Path from typing import Any +from zoneinfo import ZoneInfo from common import format_modelhub_datetime, parse_datetime, read_json, runtime_instance_id, utc_now, write_json from defaults import EMBEDDED_MODELHUB_XC_TOKEN @@ -310,6 +311,17 @@ class ModelHubClient: return active_count + def count_active_tasks_exact(self) -> int: + """Use the platform's filtered totals instead of paging all history.""" + total = 0 + for status in ("waiting", "running"): + page = self.list_tasks_page(current=1, page_size=1, only_mine=True, status=status) + data = page.get("data") + if not isinstance(data, dict) or "total" not in data: + raise ModelHubAPIError(f"Active-task total is unavailable for status={status}") + total += max(0, int(data["total"])) + return total + def _request( self, method: str, @@ -409,8 +421,6 @@ DEFAULT_CAPACITY_STATE_PATH = Path(".modelhub_state/account_capacity.json") def is_capacity_error(error: ModelHubAPIError) -> bool: - if error.code in {409, 429}: - return True message = str(error).strip().lower() if isinstance(error.payload, dict): message = f"{message} {error.payload.get('message') or ''}".lower() @@ -447,7 +457,7 @@ class ModelHubClientPool: active_counts_ttl: float | None = None, reservation_ttl: float | None = None, instance_id: str | None = None, - capacity_probe_interval_cycles: int = 3, + capacity_probe_interval_cycles: int = 0, capacity_probe_cooldown_cycles: int = 3, capacity_state_path: Path | str | None = None, recent_model_reserve_slots: int | None = None, @@ -472,7 +482,13 @@ class ModelHubClientPool: self.recent_model_days = max(1, int(configured_recent_days)) self._capacity_state_path = Path(capacity_state_path) if capacity_state_path else None self._account_keys = [self._account_key(client, index) for index, client in enumerate(clients)] + self._weekly_calibration_weeks: list[str | None] = [None for _ in clients] + self._weekly_calibration_results: list[str | None] = [None for _ in clients] self._account_caps = self._load_account_caps(self.active_task_cap) + self._weekly_calibration_week: str | None = None + self._weekly_calibration_window = False + self._weekly_calibration_deadline = 0.0 + self._weekly_no_candidate_retry_at = 0.0 self._capacity_probe_interval_cycles = max(0, int(capacity_probe_interval_cycles)) self._capacity_probe_cooldown_cycles = max(1, int(capacity_probe_cooldown_cycles)) self._capacity_probe_cycle = 0 @@ -521,10 +537,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 +555,12 @@ class ModelHubClientPool: "version": 1, "updatedAt": datetime.now().astimezone().isoformat(), "accounts": { - key: {"accountIndex": index + 1, "knownCap": self._account_caps[index]} + key: { + "accountIndex": index + 1, + "knownCap": self._account_caps[index], + "calibrationWeek": self._weekly_calibration_weeks[index], + "calibrationResult": self._weekly_calibration_results[index], + } for index, key in enumerate(self._account_keys) }, } @@ -552,6 +578,92 @@ class ModelHubClientPool: and cycle_number % self._capacity_probe_interval_cycles == 0 ) + def configure_weekly_calibration(self, now: datetime) -> dict[str, Any]: + """Only Monday 10:00-10:59 in Beijing may borrow recent-model slots.""" + local_now = now.astimezone(ZoneInfo("Asia/Shanghai")) + week = f"{local_now.isocalendar().year}-W{local_now.isocalendar().week:02d}" + window = local_now.weekday() == 0 and local_now.hour == 10 + with self._state_lock: + if week != self._weekly_calibration_week: + self._weekly_no_candidate_retry_at = 0.0 + self._weekly_calibration_week = week if window else None + self._weekly_calibration_window = window + self._weekly_calibration_deadline = ( + time.monotonic() + + (local_now.replace(hour=11, minute=0, second=0, microsecond=0) - local_now).total_seconds() + if window else 0.0 + ) + target = self._weekly_target_locked() + return { + "window": window, + "week": week, + "targetAccount": target + 1 if target is not None else None, + "completedAccounts": sum(value == week for value in self._weekly_calibration_weeks), + } + + def _weekly_target_locked(self) -> int | None: + if ( + not self._weekly_calibration_window + or not self._weekly_calibration_week + or time.monotonic() >= self._weekly_calibration_deadline + or time.monotonic() < self._weekly_no_candidate_retry_at + ): + return None + for index in range(len(self.clients)): + if ( + self._weekly_calibration_weeks[index] != self._weekly_calibration_week + and self._count_known[index] + ): + return index + return None + + def weekly_calibration_active(self) -> bool: + self._refresh_active_counts() + with self._state_lock: + return self._weekly_target_locked() is not None + + def weekly_calibration_status(self) -> dict[str, Any]: + self._refresh_active_counts() + with self._state_lock: + target = self._weekly_target_locked() + return { + "window": self._weekly_calibration_window and time.monotonic() < self._weekly_calibration_deadline, + "week": self._weekly_calibration_week, + "targetAccount": target + 1 if target is not None else None, + "completedAccounts": sum( + value == self._weekly_calibration_week + for value in self._weekly_calibration_weeks + ) if self._weekly_calibration_week else 0, + } + + def finish_weekly_calibration(self, index: int, *, result: str) -> None: + with self._state_lock: + if not self._weekly_calibration_window or not self._weekly_calibration_week: + return + self._weekly_calibration_weeks[index] = self._weekly_calibration_week + self._weekly_calibration_results[index] = result + known_cap = self._account_caps[index] + active = self._effective_count_locked(index) + threshold = self._old_model_count_limit_locked(index) + self._persist_account_caps() + print( + f"[capacity-calibration] account={index + 1:02d} result={result} " + f"known_cap={known_cap} active={active} old_threshold={threshold}", + flush=True, + ) + + def pause_weekly_calibration_for_no_candidates(self) -> None: + with self._state_lock: + target = self._weekly_target_locked() + if target is None: + return + self._weekly_no_candidate_retry_at = time.monotonic() + 600 + print( + f"[capacity-calibration] account={target + 1:02d} " + "status=paused reason=no_candidates retry_seconds=600", + flush=True, + ) + def account_capacity_limits(self) -> list[int]: with self._state_lock: return list(self._account_caps) @@ -576,8 +688,10 @@ class ModelHubClientPool: with self._state_lock: return self._capacity_probe_enabled - def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int) -> int | None: + def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int | None) -> int | None: try: + if max_count is None and hasattr(client, "count_active_tasks_exact"): + return client.count_active_tasks_exact() return client.count_active_tasks(max_count=max_count, page_size=200) except Exception: return None @@ -606,7 +720,25 @@ class ModelHubClientPool: return with self._state_lock: - count_limits = [cap + 1 for cap in self._account_caps] + count_limits: list[int | None] = [cap + 1 for cap in self._account_caps] + weekly_target = self._weekly_target_locked() + if ( + weekly_target is None + and self._weekly_calibration_window + and time.monotonic() < self._weekly_calibration_deadline + and time.monotonic() >= self._weekly_no_candidate_retry_at + ): + weekly_target = next( + ( + index for index, week in enumerate(self._weekly_calibration_weeks) + if week != self._weekly_calibration_week + ), + None, + ) + if weekly_target is not None: + # A cap+1 early stop only proves a lower bound. Weekly + # calibration needs an exact active count for its target. + count_limits[weekly_target] = None def _to_indexed_result(index: int, client: ModelHubClient) -> tuple[int, int | None]: return index, self._safe_count_active_tasks(client, count_limits[index]) @@ -694,6 +826,10 @@ class ModelHubClientPool: def available_submit_slots(self) -> int: self._refresh_active_counts() with self._state_lock: + if self._weekly_target_locked() is not None: + # The weekly run may need to cross the previously known cap. + # Bound each scan; it can resume in the next cycle. + return 100 normal_slots = sum( max(0, self._account_caps[index] - self._effective_count_locked(index)) for index in range(len(self.clients)) @@ -781,10 +917,12 @@ class ModelHubClientPool: reserve_capacity_slots: int | None = None, ) -> tuple[int, int] | None: with self._state_lock: + weekly_target = self._weekly_target_locked() remaining_by_index = { index: self._account_caps[index] - self._effective_count_locked(index) for index in range(len(self.clients)) if index not in excluded + and (weekly_target is None or index == weekly_target) and ( reserve_capacity_slots is None or ( @@ -819,14 +957,19 @@ class ModelHubClientPool: reserve_capacity_slots: int | None = None, ) -> tuple[int, int] | None: with self._state_lock: - if not self._capacity_probe_enabled: + weekly_target = self._weekly_target_locked() + if weekly_target is None and not self._capacity_probe_enabled: return None eligible = [ index for index in range(len(self.clients)) if index not in excluded - and index not in self._capacity_probe_attempted - and 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 +987,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] = { @@ -877,7 +1021,12 @@ class ModelHubClientPool: if observed_capacity < self._account_caps[index]: self._account_caps[index] = observed_capacity cap_changed = True - self._remote_counts[index] = self._account_caps[index] + # Accepted local reservations may not have appeared in the last + # remote count yet. Do not count them twice when marking full. + self._remote_counts[index] = max( + 0, + self._account_caps[index] - len(self._reservations[index]), + ) self._count_known[index] = True self._counts_initialized = True self._active_refresh_at = time.monotonic() @@ -890,14 +1039,16 @@ class ModelHubClientPool: if discovered_cap <= self._account_caps[index]: return self._account_caps[index] = discovered_cap + weekly = self._weekly_target_locked() == index self._persist_account_caps() - 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 +1056,7 @@ class ModelHubClientPool: for index in range(len(self.clients)) if self._count_known[index] ) + return max(1, normal_slots) if self._weekly_target_locked() is not None else normal_slots def _old_model_count_limit_locked(self, index: int) -> int: return max(0, self._account_caps[index] - self.recent_model_reserve_slots) @@ -943,9 +1095,15 @@ class ModelHubClientPool: attempted_accounts: set[int] = set() forced_refresh_done = False last_capacity_error: ModelHubAPIError | None = None - reserved_capacity_slots = self.recent_model_reserve_slots if old_model_only else None while True: + with self._state_lock: + weekly_target = self._weekly_target_locked() + reserved_capacity_slots = ( + self.recent_model_reserve_slots + if old_model_only and weekly_target is None + else None + ) reservation = self._reserve_account( attempted_accounts, reserve_capacity_slots=reserved_capacity_slots, @@ -977,6 +1135,9 @@ class ModelHubClientPool: selected_index, reservation_id = reservation selected_client = self.clients[selected_index] + if weekly_target == selected_index and time.monotonic() >= self._weekly_calibration_deadline: + self._finish_reservation(selected_index, reservation_id, succeeded=False) + continue try: response = selected_client.add_task(payload) except ModelHubAPIError as exc: @@ -986,6 +1147,18 @@ class ModelHubClientPool: # Another process may have filled this account after our count # refresh. Mark it full locally and immediately try another one. self._mark_account_saturated(selected_index, capacity_probe=capacity_probe) + if weekly_target == selected_index: + self._refresh_active_counts(force=True) + with self._state_lock: + confirmed = ( + self._count_known[selected_index] + and self._remote_counts[selected_index] >= self._account_caps[selected_index] + ) + self.finish_weekly_calibration( + selected_index, + result="saturated" if confirmed else "inconclusive_capacity_response", + ) + self._refresh_active_counts(force=True) attempted_accounts.add(selected_index) last_capacity_error = exc continue diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index acff228d8d..44744eabf1 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -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']} " diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 5f567de2a9..c2eb688126 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.09.20.1" +AGENT_VERSION = "2026.09.20.2" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 2ccbf8b11d..ef42107b8b 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -20,6 +20,7 @@ from common import read_json, read_jsonl # noqa: E402 from main import build_adaptive_scan_stages, build_parser, make_run_dir, run_submission, submit_candidate # noqa: E402 from modelhub_client import ( # noqa: E402 ModelHubAPIError, + ModelHubClient, ModelHubClientPool, OldModelQueuePolicyError, is_duplicate_submission_error, @@ -120,8 +121,8 @@ class DynamicCapacityClient: self.limit = limit self.token = token - def count_active_tasks(self, *, max_count: int, **_kwargs) -> int: # noqa: ANN003 - 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 +216,195 @@ def make_candidate(index: int) -> dict: class ClientPoolConcurrencyTests(unittest.TestCase): + def test_exact_active_count_uses_filtered_platform_totals(self) -> None: + client = ModelHubClient(token="test-token") + calls: list[str] = [] + + def page(*, status: str, **_kwargs) -> dict: # noqa: ANN003 + calls.append(status) + return {"data": {"total": {"waiting": 175, "running": 25}[status]}} + + client.list_tasks_page = page # type: ignore[method-assign] + self.assertEqual(200, client.count_active_tasks_exact()) + self.assertEqual(["waiting", "running"], calls) + + @staticmethod + def monday_calibration_time(hour: int = 10) -> datetime: + # 2026-09-21 10:00 Beijing time. + return datetime(2026, 9, 21, hour - 8, tzinfo=timezone.utc) + + def test_weekly_calibration_borrows_five_slots_only_during_monday_window(self) -> None: + client = DynamicCapacityClient(active=95, limit=100) + pool = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + recent_model_reserve_slots=5, + ) + self.assertEqual([95], pool.active_task_counts()) + old_time = datetime(2026, 9, 1, tzinfo=timezone.utc) + pool.configure_weekly_calibration(self.monday_calibration_time(9)) + with self.assertRaises(OldModelQueuePolicyError): + pool.add_task_for_model({"model": "before"}, model_last_modified=old_time) + + pool.configure_weekly_calibration(self.monday_calibration_time()) + self.assertEqual(1, pool.old_model_submit_slots()) + for index in range(5): + pool.add_task_for_model({"model": index}, model_last_modified=old_time) + self.assertEqual(100, client.active) + with self.assertRaises(ModelHubAPIError): + pool.add_task_for_model({"model": "test-101"}, model_last_modified=old_time) + self.assertEqual(1, pool.weekly_calibration_status()["completedAccounts"]) + self.assertEqual([95], pool.old_model_queue_thresholds()) + pool.configure_weekly_calibration(self.monday_calibration_time(11)) + with self.assertRaises(OldModelQueuePolicyError): + pool.add_task_for_model({"model": "after"}, model_last_modified=old_time) + self.assertEqual(100, client.active) + + def test_weekly_calibration_stops_when_window_deadline_passes(self) -> None: + client = DynamicCapacityClient(active=95, limit=200) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + pool.active_task_counts() + pool.configure_weekly_calibration(self.monday_calibration_time()) + pool._weekly_calibration_deadline = 0.0 + with self.assertRaises(OldModelQueuePolicyError): + pool.add_task_for_model( + {"model": "too-late"}, + model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc), + ) + self.assertEqual(95, client.active) + + def test_weekly_calibration_discovers_200_and_persists_per_account_completion(self) -> None: + client = DynamicCapacityClient(active=95, limit=200) + old_time = datetime(2026, 9, 1, tzinfo=timezone.utc) + with tempfile.TemporaryDirectory() as temporary_dir: + state_path = Path(temporary_dir) / "account_capacity.json" + pool = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + capacity_state_path=state_path, + ) + pool.active_task_counts() + pool.configure_weekly_calibration(self.monday_calibration_time()) + for index in range(105): + pool.add_task_for_model({"model": index}, model_last_modified=old_time) + with self.assertRaises(ModelHubAPIError): + pool.add_task_for_model({"model": "test-201"}, model_last_modified=old_time) + self.assertEqual([200], pool.account_capacity_limits()) + self.assertEqual([195], pool.old_model_queue_thresholds()) + stored = read_json(state_path) + account = next(iter(stored["accounts"].values())) + self.assertEqual("saturated", account["calibrationResult"]) + + restarted = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + capacity_state_path=state_path, + ) + restarted.active_task_counts() + restarted.configure_weekly_calibration(self.monday_calibration_time()) + self.assertEqual([200], restarted.account_capacity_limits()) + self.assertIsNone(restarted.weekly_calibration_status()["targetAccount"]) + + def test_weekly_calibration_reads_full_active_count_not_cap_plus_one(self) -> None: + client = DynamicCapacityClient(active=200, limit=200) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + pool.configure_weekly_calibration(self.monday_calibration_time()) + self.assertEqual([200], pool.active_task_counts()) + self.assertEqual([200], pool.account_capacity_limits()) + with self.assertRaises(ModelHubAPIError): + pool.add_task_for_model( + {"model": "full-capacity"}, + model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc), + ) + self.assertEqual(1, pool.weekly_calibration_status()["completedAccounts"]) + + def test_weekly_calibration_targets_accounts_separately(self) -> None: + first = DynamicCapacityClient(active=2, limit=2, token="first") + second = DynamicCapacityClient(active=1, limit=3, token="second") + pool = ModelHubClientPool( + [first, second], # type: ignore[list-item] + active_task_cap=2, + recent_model_reserve_slots=1, + ) + pool.active_task_counts() + pool.configure_weekly_calibration(self.monday_calibration_time()) + self.assertEqual(1, pool.weekly_calibration_status()["targetAccount"]) + pool.add_task_for_model( + {"model": "try-first-then-second"}, + model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc), + ) + self.assertEqual(2, first.active) + self.assertEqual(2, second.active) + self.assertEqual(2, pool.weekly_calibration_status()["targetAccount"]) + pool.add_task_for_model( + {"model": "second-capacity"}, + model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc), + ) + with self.assertRaises(ModelHubAPIError): + pool.add_task_for_model( + {"model": "test-second-full"}, + model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc), + ) + self.assertEqual([2, 3], pool.account_capacity_limits()) + self.assertEqual([1, 2], pool.old_model_queue_thresholds()) + self.assertEqual(2, pool.weekly_calibration_status()["completedAccounts"]) + + def test_weekly_calibration_skips_unknown_counts_and_retries_no_candidates(self) -> None: + class UnreadableCountClient(DynamicCapacityClient): + def count_active_tasks(self, **_kwargs) -> int: + raise TimeoutError("cannot read active count") + + unreadable = UnreadableCountClient(active=95, limit=100, token="unreadable") + readable = DynamicCapacityClient(active=95, limit=100, token="readable") + pool = ModelHubClientPool( + [unreadable, readable], # type: ignore[list-item] + active_task_cap=100, + ) + pool.active_task_counts() + pool.configure_weekly_calibration(self.monday_calibration_time()) + self.assertEqual(2, pool.weekly_calibration_status()["targetAccount"]) + pool.pause_weekly_calibration_for_no_candidates() + self.assertEqual(0, pool.weekly_calibration_status()["completedAccounts"]) + self.assertIsNone(pool.weekly_calibration_status()["targetAccount"]) + pool._weekly_no_candidate_retry_at = 0.0 + self.assertEqual(2, pool.weekly_calibration_status()["targetAccount"]) + + def test_rate_limit_does_not_finish_weekly_capacity_calibration(self) -> None: + class RateLimitedClient(DynamicCapacityClient): + def add_task(self, _payload: dict) -> dict: + raise ModelHubAPIError("rate limited", code=429) + + client = RateLimitedClient(active=100, limit=200) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + pool.active_task_counts() + pool.configure_weekly_calibration(self.monday_calibration_time()) + with self.assertRaises(ModelHubAPIError): + pool.add_task_for_model( + {"model": "rate-limited"}, + model_last_modified=datetime(2026, 9, 1, tzinfo=timezone.utc), + ) + self.assertEqual([100], pool.account_capacity_limits()) + self.assertEqual(1, pool.weekly_calibration_status()["targetAccount"]) + + def test_rate_limited_submission_is_deferred_not_failed_or_capacity(self) -> None: + class RateLimitedClient: + @staticmethod + def add_task(_payload: dict) -> dict: + raise ModelHubAPIError("rate limited", code=429) + + result = submit_candidate( + { + "repoId": "owner/model", + "modelAddress": "https://modelscope.cn/models/owner/model", + "targetGpu": "gpu-a", + "framework": "vllm", + "taskType": "text-generation", + "configParams": "safe", + }, + RateLimitedClient(), # type: ignore[arg-type] + ) + self.assertEqual("rate_limit_deferred", result["outcome"]) + def test_old_models_use_another_account_and_reserve_last_five_positions(self) -> None: below_threshold = FakeClient(active_count=94) at_threshold = FakeClient(active_count=95)