From 91d1d3d87d5995a75542771fdf835c60fd9a3123 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Wed, 12 Aug 2026 10:18:09 +0800 Subject: [PATCH] feat: make model age admission-only --- README.md | 57 +++-- modelhub_submmit_api/README.md | 35 +-- modelhub_submmit_api/daily_runner.py | 4 +- modelhub_submmit_api/gpu_strategy.py | 14 +- modelhub_submmit_api/main.py | 73 +++--- modelhub_submmit_api/modelhub_client.py | 45 ++-- modelhub_submmit_api/outcome_tracker.py | 54 ++++- modelhub_submmit_api/poll_runner.py | 53 +---- modelhub_submmit_api/queue_cleanup.py | 284 ++---------------------- modelhub_submmit_api/version.py | 2 +- tests/test_candidate_preflight.py | 44 ++++ tests/test_concurrency.py | 135 +++++++++-- tests/test_gpu_strategy.py | 28 +++ tests/test_poll_policy.py | 31 +-- tests/test_queue_cleanup.py | 232 ++----------------- 15 files changed, 438 insertions(+), 653 deletions(-) diff --git a/README.md b/README.md index f1e9ef0d..075f0aba 100644 --- a/README.md +++ b/README.md @@ -62,8 +62,7 @@ Optional tuning: - `MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT` default `5000` - `MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS` default `8` - `MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS` default `0` (unlimited) -- `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` default `10` per account -- `MODELHUB_DYNAMIC_OLD_MODEL_CLEANUP_RESERVE_SLOTS` default `5` per account +- `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` default `5` per account - `MODELHUB_RECENT_MODEL_DAYS` default `7` ## Adaptive GPU Strategy @@ -171,37 +170,45 @@ same model or infer failure from historical similarity. The cleanup repeats every 120 poll cycles by default and writes its full evidence report to `.modelhub_state/queue_cleanup_latest.json`. -Architecture cleanup joins each active task to the locally recorded submission -or ledger entry to recover its exact framework and task type, then reads the -model's `config.json`. Only an exact GPU + framework + task type + architecture -blacklist hit can authorize cancellation. Matching waiting tasks are stopped; +Architecture cleanup reads the live queue task level and joins local submission +or ledger metadata when available, then reads the model's `config.json`. An +exact GPU + framework + task type + architecture blacklist hit can authorize +cancellation. For legacy waiting tasks whose framework is absent from the API +and local state, every currently listed framework must have an explicit matching +block; a partial match is never enough. Matching waiting tasks are stopped; running tasks remain protected and their state is rechecked again immediately before the stop call. Failed outcomes are synchronized every three poll cycles. When fixed `MODEL_NOT_SUPPORTED` text adds a new blacklist entry, a lightweight architecture-only cleanup runs immediately without repeating repository-size or model-age scans. -Each account dynamically reserves its last 10 known-capacity positions for +Each account dynamically reserves its last 5 known-capacity positions for models updated within seven days. If an account's discovered limit is 100, 200, -or 500, older models stop at positions 90, 190, or 490 respectively. Old-model +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. +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. -On startup, the worker first stops deterministic OOM tasks, recalculates each -account's surviving task order by numeric task ID, and removes -older-than-seven-days tasks beyond that account's current limit minus 10. Later -scheduled cleanup uses limit minus 5, retaining a small hysteresis buffer that -avoids repeatedly over-pruning valid work. For a 100-task account the two -boundaries are 90 and 95; for a 500-task account they are 490 and 495. Recent -overflow tasks are always kept. Age-based cleanup applies only to tasks that are -still waiting; running validation tasks are protected even beyond the boundary. -ModelScope metadata failures fail closed and never trigger cancellation. -Immediately before the age-only stop batch, task ownership, active status, and -post-OOM queue position are checked again. Deterministic OOM cleanup may still -stop a running task because it cannot fit the selected GPU. +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 +tasks remain untouched by date policy. Cleanup still removes deterministic OOM +tasks and exact learned architecture incompatibilities under their existing +state and evidence checks, preventing the old submit/cleanup/resubmit loop. + +A full 12-account failure audit found that 31.25% of failures had no usable log +and another 23.52% had only ambiguous runtime evidence. These unresolved results +now remain visible as `unresolvedFailureCount` but do not reduce GPU/framework +decision success rates, create model/GPU cooldowns, or open attributable-failure +circuits. Explicit OOM, repository, context, tokenizer, model-load, operator, and +framework-architecture evidence remains attributable; identified platform +failures continue to use short infrastructure circuits instead. The verified capacities, safe repository-size boundaries, evidence hierarchy, and source links are recorded in @@ -311,12 +318,18 @@ Version `2026.08.12.5` enriches every cleanup from the live active queue's task level metadata. If a legacy waiting task has no recoverable framework, cleanup queries the current ModelHub framework catalog and cancels it only when its architecture is explicitly blocked on every listed framework. +Version `2026.08.12.6` makes model age admission-only: each account reserves its +last five dynamic-capacity positions for seven-day models, unknown account +counts fail closed for older candidates, and no startup or periodic cleanup can +cancel a task by date. It also keeps unclassified/ambiguous historical failures +neutral in GPU/framework success feedback while preserving deterministic OOM +and architecture cleanup. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag agent-v23 -git push origin agent-v23 +git tag agent-v24 +git push origin agent-v24 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 82b98190..12e7d7e0 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -16,7 +16,7 @@ It currently supports: - `main.py`: core discovery, scoring, dedup, and submission - `daily_runner.py`: daily wave orchestration - `poll_runner.py`: long-running queue refiller -- `queue_cleanup.py`: fail-closed cleanup for certain OOM, architecture, and age policies +- `queue_cleanup.py`: fail-closed cleanup for deterministic OOM and architecture policies - `runner_common.py`: shared token / key file loading - `hf_discovery.py`: ModelScope model discovery and inspection (keeps the legacy module name) - `modelhub_client.py`: ModelHub API client and token-pool routing @@ -137,14 +137,13 @@ bash run_poll.sh --dry-run memory rule as new submissions. Only tasks whose own repository size times `1.20` exceeds their selected GPU capacity are stopped, after a fresh account-scoped active-state check. Incomplete size/capacity evidence is never used for cancellation. -- Models older than seven days may occupy only the current account capacity minus its final 10 - positions. The account pool enforces this per-account boundary atomically and updates it when - capacity probing discovers a higher limit. Cleanup stops OOM tasks first, recalculates the - surviving queue order, and applies the same limit-minus-10 boundary on startup. Scheduled - cleanup relaxes to limit minus 5 to avoid excessive pruning. Age cleanup stops waiting tasks - only; running tasks are protected and rechecked after OOM cleanup, immediately before the - age-only stop batch. Recent - overflow tasks stay, and unknown ModelScope timestamps never authorize a 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 + 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. + Date-deferred candidates are skipped, not failed or permanently excluded. ## Important Flags @@ -178,19 +177,25 @@ Common flags: - `--submit-concurrency`: concurrent task submission calls used by each cycle (0 = auto) - `--post-cycle-cooldown-seconds`: pause after a successful cycle before next cycle (default 2) - `--max-cycles`: optional hard stop for testing or batch windows -- `--disable-queue-cleanup`: disable automatic OOM and old-overflow queue cleanup +- `--disable-queue-cleanup`: disable automatic deterministic OOM/architecture queue cleanup -The first automatic cleanup removes models older than seven days beyond each -account's discovered capacity minus 10. Later scheduled cleanups use capacity -minus 5, while admission continues to reserve the final 10 slots for recent -models. Override these suffix sizes with `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` -and `MODELHUB_DYNAMIC_OLD_MODEL_CLEANUP_RESERVE_SLOTS`. +Model age is admission-only. The default `MODELHUB_RECENT_MODEL_RESERVE_SLOTS=5` +reserves each account's final five known-capacity positions for models updated +within seven days. Queue cleanup reports `ageCleanupMode=admission_only` and the +compatibility field `oldOverflowCount=0`; no restart migration or periodic +date-based stop is performed. Every successful worker-initiated stop is persisted as `policy_cancelled` in the outcome store. It is excluded from GPU/framework success rates, local failure cooldowns, and circuit breakers. If a task races to a real success before the stop takes effect, that success remains authoritative. +Unclassified failures and failures with `failureScope=unknown` are reported as +`unresolvedFailureCount`. They do not penalize GPU/framework decision success, +open attributable-failure circuits, or create model/GPU cooldowns. Explicitly +classified non-platform failures remain strategy evidence, while platform +failures retain their separate short-circuit handling. + Failure-informed preflight is enabled by default. It rejects deterministic missing-file and predicted-OOM cases, clamps unsafe context-length arguments, and records its decisions in `candidatePreflight` and each candidate's diff --git a/modelhub_submmit_api/daily_runner.py b/modelhub_submmit_api/daily_runner.py index 7f6a1af8..e46cf298 100644 --- a/modelhub_submmit_api/daily_runner.py +++ b/modelhub_submmit_api/daily_runner.py @@ -88,7 +88,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--recent-model-reserve-slots", type=int, - default=int(os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "10")), + default=int(os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "5")), help=argparse.SUPPRESS, ) parser.add_argument( @@ -252,7 +252,7 @@ def make_wave_namespace(base_args: argparse.Namespace, wave: WaveSpec) -> argpar capacity_probe_interval_cycles=getattr(base_args, "capacity_probe_interval_cycles", 3), submit_concurrency=getattr(base_args, "submit_concurrency", 1), max_submits_per_run=getattr(base_args, "max_submits_per_run", 0), - recent_model_reserve_slots=getattr(base_args, "recent_model_reserve_slots", 10), + recent_model_reserve_slots=getattr(base_args, "recent_model_reserve_slots", 5), recent_model_days=getattr(base_args, "recent_model_days", 7), disable_candidate_preflight=getattr(base_args, "disable_candidate_preflight", False), llm_classifier_endpoint=getattr(base_args, "llm_classifier_endpoint", None), diff --git a/modelhub_submmit_api/gpu_strategy.py b/modelhub_submmit_api/gpu_strategy.py index 4633e08f..ee132249 100644 --- a/modelhub_submmit_api/gpu_strategy.py +++ b/modelhub_submmit_api/gpu_strategy.py @@ -17,7 +17,7 @@ DEFAULT_GPU_STRATEGY_PATH = Path(".modelhub_state/gpu_strategy.json") DEFAULT_REFRESH_SUBMISSIONS = 200 DEFAULT_RECENT_TERMINAL_WINDOW = 1000 DEFAULT_LONG_TERM_MIN_SAMPLES = 100 -STRATEGY_STATE_VERSION = 3 +STRATEGY_STATE_VERSION = 4 LONG_TERM = "long_term" RECENT = "recent" @@ -291,6 +291,7 @@ class GPUStrategyManager: *, supported_gpus: list[str], now: datetime | None = None, + history_records: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: now = now or utc_now() supported_gpus = list(dict.fromkeys(gpu for gpu in supported_gpus if gpu)) @@ -304,7 +305,11 @@ class GPUStrategyManager: previous_generation = int(state.get("generation", -1)) if state is not None else -1 self.log(f"[strategy] refresh_start reason={reason} supported_gpus={len(supported_gpus)}") try: - tasks = self._load_platform_history(client) + tasks = ( + list(history_records) + if history_records is not None + else self._load_platform_history(client) + ) refreshed = build_strategy_snapshot( tasks, supported_gpus=supported_gpus, @@ -315,6 +320,11 @@ class GPUStrategyManager: refresh_submissions=self.refresh_submissions, ) refreshed["acceptedTotal"] = int((state or {}).get("acceptedTotal") or 0) + refreshed["historySource"] = ( + "classified_attributable_outcomes" + if history_records is not None + else "raw_platform_history" + ) self.state = refreshed write_json(self.path, refreshed) self._log_state("refreshed") diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index b78cb5c5..4ef6129a 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -111,7 +111,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--recent-model-reserve-slots", type=int, - default=int(os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "10")), + default=int(os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "5")), help=argparse.SUPPRESS, ) parser.add_argument( @@ -749,10 +749,15 @@ def submit_candidate( } except ModelHubAPIError as exc: if isinstance(exc, OldModelQueuePolicyError): + print( + f"[submit] deferred repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=age_policy_skipped", + flush=True, + ) return { - "outcome": "old_model_policy_skipped", + "outcome": "age_policy_deferred", "candidate": candidate, - "reason": "all_accounts_at_old_model_queue_threshold", + "reason": "age_policy_skipped", } if is_model_uniqueness_error(exc): print( @@ -839,7 +844,7 @@ def run_submission( clients, capacity_probe_interval_cycles=max(0, int(getattr(args, "capacity_probe_interval_cycles", 3) or 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", 10) or 0)), + 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)), ) @@ -1054,7 +1059,13 @@ def run_submission( long_term_min_samples=max(1, int(getattr(args, "gpu_strategy_min_long_samples", 100) or 100)), market_intelligence=market_intelligence, ) - strategy_manager.prepare(modelhub_client, supported_gpus=target_gpus, now=now) + strategy_history_records = outcome_tracker.get_strategy_history_records() + strategy_manager.prepare( + modelhub_client, + supported_gpus=target_gpus, + now=now, + history_records=strategy_history_records, + ) strategy_summary = strategy_manager.summary() if getattr(args, "skip_history_archive", False): @@ -1108,17 +1119,18 @@ def run_submission( pipeline_tags = pipeline_tags_for_task_types(selected_task_types) explicit_scan_cap = max(0, int(getattr(args, "max_scan_models", 0) or 0)) recent_model_days = max(1, int(getattr(args, "recent_model_days", 7) or 7)) - recent_model_reserve_slots = max(0, int(getattr(args, "recent_model_reserve_slots", 10) or 0)) + recent_model_reserve_slots = max(0, int(getattr(args, "recent_model_reserve_slots", 5) or 0)) old_model_slots_before_scan: int | None = None if hasattr(modelhub_client, "old_model_submit_slots"): old_model_slots_before_scan = int(modelhub_client.old_model_submit_slots()) - old_model_queue_thresholds: list[int] | None = None + 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()) allow_older_models_for_scan = old_model_slots_before_scan is None or old_model_slots_before_scan > 0 print( - f"[age-policy] recent_days={recent_model_days} reserve_recent_slots={recent_model_reserve_slots} " - f"account_thresholds={','.join(str(value) for value in old_model_queue_thresholds) if old_model_queue_thresholds is not None else 'n/a'} " + f"[age-policy] mode=admission_only recent_days={recent_model_days} " + f"reserve_recent_slots={recent_model_reserve_slots} " + 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'}", flush=True, @@ -1133,6 +1145,7 @@ def run_submission( ): if candidate_goal <= 0 or len(candidates) >= candidate_goal: break + skipped_before_stage = len(skipped) stage_updated_after = stage["updatedAfter"] stage_limit = int(stage["limit"]) query_kwargs = { @@ -1149,15 +1162,26 @@ def run_submission( except TypeError: models = hf_discovery.list_recent_models(**query_kwargs) if not allow_older_models_for_scan: - models = [ - model - for model in models + recent_models: list[HFModelSummary] = [] + for model in models: if is_recent_model( model.last_modified, reference_time=now, recent_model_days=recent_model_days, + ): + recent_models.append(model) + continue + if model.repo_id in seen_model_ids: + continue + seen_model_ids.add(model.repo_id) + skipped.append( + { + "repoId": model.repo_id, + "reason": "age_policy_skipped", + "deferred": True, + } ) - ] + models = recent_models stage_candidates, stage_skipped, stage_failed, processed_count = collect_candidates_from_models( models=models, @@ -1186,14 +1210,14 @@ def run_submission( "newModelsProcessed": processed_count, "candidatesAdded": len(stage_candidates), "candidateCountAfterStage": len(candidates), - "skippedAdded": len(stage_skipped), + "skippedAdded": len(skipped) - skipped_before_stage, "failedAdded": len(stage_failed), } scan_stages.append(stage_summary) print( f"[scan] stage={stage['name']} discovered={len(models)} new_processed={processed_count} " f"candidates_added={len(stage_candidates)} candidates_total={len(candidates)}/{candidate_goal} " - f"skipped_added={len(stage_skipped)}", + f"skipped_added={len(skipped) - skipped_before_stage}", flush=True, ) @@ -1235,25 +1259,11 @@ def run_submission( # already-scanned pool until the desired number of real submissions is # reached or account capacity is genuinely exhausted. while len(submitted) < target_submit_count and len(attempted_candidates) < max_submit_attempts: - allow_older_models_now = ( - bool(modelhub_client.can_submit_old_models()) - if hasattr(modelhub_client, "can_submit_old_models") - else True - ) - submission_reference_time = utc_now() remaining_candidates = [ candidate for candidate in diversified_candidates if candidate_key(candidate) not in attempted_keys and not submission_exclusion_store.is_blocked(candidate["repoId"], candidate["targetGpu"]) - and ( - allow_older_models_now - or is_recent_model( - candidate.get("lastModified"), - reference_time=submission_reference_time, - recent_model_days=recent_model_days, - ) - ) ] remaining_candidates = one_candidate_per_model(remaining_candidates) desired_count = min( @@ -1328,13 +1338,13 @@ def run_submission( } ) continue - if result["outcome"] == "old_model_policy_skipped": + if result["outcome"] == "age_policy_deferred": batch_policy_skipped_candidates.append(candidate) skipped.append( { "repoId": candidate["repoId"], "targetGpu": candidate["targetGpu"], - "reason": "all_accounts_at_old_model_queue_threshold", + "reason": "age_policy_skipped", } ) continue @@ -1428,6 +1438,7 @@ def run_submission( "marketIntelligence": market_summary, "candidatePreflight": preflight_summary, "agePolicy": { + "mode": "admission_only", "recentModelDays": recent_model_days, "recentModelReserveSlots": recent_model_reserve_slots, "oldModelSubmitThresholds": old_model_queue_thresholds, diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 6d329a0d..238567b3 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -407,7 +407,7 @@ class ModelHubClientPool: configured_recent_reserve = ( recent_model_reserve_slots if recent_model_reserve_slots is not None - else os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "10") + else os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "5") ) configured_recent_days = ( recent_model_days @@ -434,6 +434,10 @@ class ModelHubClientPool: self._active_counts_ttl = max(1.0, float(configured_ttl)) self._reservation_ttl = max(self._active_counts_ttl * 2, float(configured_reservation_ttl)) self._remote_counts: list[int] = [0 for _ in clients] + # Old-model admission is fail-closed per account. A configured or + # persisted capacity is not enough without a successful active-count + # read for the current scheduling window. + self._count_known: list[bool] = [False for _ in clients] self._active_refresh_at: float = 0.0 self._counts_initialized = False self._state_lock = threading.Lock() @@ -518,11 +522,11 @@ class ModelHubClientPool: with self._state_lock: return self._capacity_probe_enabled - def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int) -> int: + def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int) -> int | None: try: return client.count_active_tasks(max_count=max_count, page_size=200) except Exception: - return max(0, max_count - 1) + return None def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]: try: @@ -550,10 +554,10 @@ class ModelHubClientPool: with self._state_lock: count_limits = [cap + 1 for cap in self._account_caps] - def _to_indexed_result(index: int, client: ModelHubClient) -> tuple[int, int]: + def _to_indexed_result(index: int, client: ModelHubClient) -> tuple[int, int | None]: return index, self._safe_count_active_tasks(client, count_limits[index]) - results: list[tuple[int, int]] = [] + results: list[tuple[int, int | None]] = [] with ThreadPoolExecutor(max_workers=min(len(self.clients), 12)) as executor: futures = { executor.submit(_to_indexed_result, index, client): index @@ -564,14 +568,18 @@ class ModelHubClientPool: try: results.append(future.result()) except Exception: - results.append((index, self.active_task_cap)) + results.append((index, None)) refreshed_at = time.monotonic() caps_changed = False with self._state_lock: for index, count in results: + if count is None: + self._count_known[index] = False + continue old_remote_count = self._remote_counts[index] new_remote_count = max(0, int(count)) + self._count_known[index] = True if new_remote_count > self._account_caps[index]: self._account_caps[index] = new_remote_count caps_changed = True @@ -725,8 +733,11 @@ class ModelHubClientPool: if index not in excluded and ( reserve_capacity_slots is None - or self._effective_count_locked(index) - < max(0, self._account_caps[index] - reserve_capacity_slots) + or ( + self._count_known[index] + and self._effective_count_locked(index) + < max(0, self._account_caps[index] - reserve_capacity_slots) + ) ) } usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0} @@ -765,8 +776,11 @@ class ModelHubClientPool: and self._effective_count_locked(index) >= self._account_caps[index] and ( reserve_capacity_slots is None - or self._effective_count_locked(index) - < max(0, self._account_caps[index] - reserve_capacity_slots) + or ( + self._count_known[index] + and self._effective_count_locked(index) + < max(0, self._account_caps[index] - reserve_capacity_slots) + ) ) ] if not eligible: @@ -810,6 +824,7 @@ class ModelHubClientPool: self._account_caps[index] = observed_capacity cap_changed = True self._remote_counts[index] = self._account_caps[index] + self._count_known[index] = True self._counts_initialized = True self._active_refresh_at = time.monotonic() if cap_changed: @@ -834,15 +849,19 @@ class ModelHubClientPool: self._old_model_count_limit_locked(index) - self._effective_count_locked(index), ) for index in range(len(self.clients)) + if self._count_known[index] ) def _old_model_count_limit_locked(self, index: int) -> int: return max(0, self._account_caps[index] - self.recent_model_reserve_slots) - def old_model_queue_thresholds(self) -> list[int]: - """Return each account's current cap minus its recent-model reserve.""" + def old_model_queue_thresholds(self) -> list[int | None]: + """Return thresholds, or None when old-model admission must fail closed.""" with self._state_lock: - return [self._old_model_count_limit_locked(index) for index in range(len(self.clients))] + return [ + self._old_model_count_limit_locked(index) if self._count_known[index] else None + for index in range(len(self.clients)) + ] def can_submit_old_models(self) -> bool: return self.old_model_submit_slots() > 0 diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index 0fcca09f..2674e5c1 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -76,6 +76,28 @@ class OutcomeTracker: } return contexts + def get_strategy_history_records(self) -> list[dict[str, Any]]: + """Expose only successes and evidence-attributable failures for GPU ranking.""" + records: list[dict[str, Any]] = [] + for record in self._records: + outcome = record.get("outcome") + if outcome == "success": + verify_result = 1 + elif _is_attributable_failure(record): + verify_result = -1 + else: + continue + records.append( + { + **record, + "gpuType": record.get("targetGpu"), + "status": "success", + "verifyResult": verify_result, + "updateTime": record.get("lastSyncTime") or record.get("submitTime"), + } + ) + return records + def _rebuild_indexes(self) -> None: self._by_task_id.clear() self._by_model_gpu.clear() @@ -619,10 +641,13 @@ class OutcomeTracker: self._failed_model_gpus.clear() latest_by_combo: dict[tuple[str, str], tuple[datetime, dict[str, Any]]] = {} for record in self._records: - # Infrastructure failures and our own policy cancellations neither - # clear nor create a model/GPU cooldown. Look through them to the - # latest attributable outcome. - if _is_platform_failure(record) or _is_policy_cancelled(record): + # Platform, unresolved, and policy outcomes neither clear nor + # create a model/GPU cooldown. The full-history audit showed that + # more than half of failures lack enough evidence for attribution. + if _is_policy_cancelled(record) or ( + record.get("outcome") == "failed" + and not _is_attributable_failure(record) + ): continue model_id = record.get("modelId") or "" target_gpu = record.get("targetGpu") or "" @@ -701,10 +726,13 @@ def _summarize(records: list[dict[str, Any]]) -> dict[str, Any]: failure_count = sum(1 for r in records if r.get("outcome") == "failed") pending_count = sum(1 for r in records if r.get("outcome") == "pending") attributable_failure_count = sum( - 1 for record in records - if record.get("outcome") == "failed" and not _is_platform_failure(record) + 1 for record in records if _is_attributable_failure(record) + ) + platform_failure_count = sum(1 for record in records if _is_platform_failure(record)) + unresolved_failure_count = max( + 0, + failure_count - attributable_failure_count - platform_failure_count, ) - platform_failure_count = failure_count - attributable_failure_count decision_total = success_count + attributable_failure_count failure_breakdown: dict[str, int] = defaultdict(int) @@ -719,6 +747,7 @@ def _summarize(records: list[dict[str, Any]]) -> dict[str, Any]: "failureCount": failure_count, "attributableFailureCount": attributable_failure_count, "platformFailureCount": platform_failure_count, + "unresolvedFailureCount": unresolved_failure_count, "decisionTotal": decision_total, "pendingCount": pending_count, "successRate": round(success_count / total, 4) if total > 0 else 0.0, @@ -892,6 +921,15 @@ def _is_platform_failure(record: dict[str, Any]) -> bool: return scope == "platform" or category.startswith("platform_") +def _is_attributable_failure(record: dict[str, Any]) -> bool: + """Require classified, non-platform evidence before penalizing a strategy.""" + if record.get("outcome") != "failed" or _is_platform_failure(record): + return False + category = str(record.get("failureCategory") or "").strip().lower() + scope = str(record.get("failureScope") or "").strip().lower() + return bool(category and scope not in {"", "unknown", "platform"}) + + def _is_policy_cancelled(record: dict[str, Any]) -> bool: return bool(record.get("policyCancelled")) or record.get("outcome") == "policy_cancelled" @@ -901,7 +939,7 @@ def _consecutive_attributable_failures(records: list[dict[str, Any]]) -> int: for record in records: if record.get("outcome") == "success": break - if record.get("outcome") == "failed" and not _is_platform_failure(record): + if _is_attributable_failure(record): count += 1 return count diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index d48ffa6e..f9e62bc1 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -74,7 +74,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument( "--recent-model-reserve-slots", type=int, - default=int(os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "10")), + default=int(os.getenv("MODELHUB_RECENT_MODEL_RESERVE_SLOTS", "5")), help=argparse.SUPPRESS, ) parser.add_argument( @@ -83,12 +83,6 @@ def build_parser() -> argparse.ArgumentParser: default=int(os.getenv("MODELHUB_RECENT_MODEL_DAYS", "7")), help=argparse.SUPPRESS, ) - parser.add_argument( - "--dynamic-old-model-cleanup-reserve-slots", - type=int, - default=int(os.getenv("MODELHUB_DYNAMIC_OLD_MODEL_CLEANUP_RESERVE_SLOTS", "5")), - help=argparse.SUPPRESS, - ) parser.add_argument("--disable-candidate-preflight", action="store_true", help=argparse.SUPPRESS) parser.add_argument("--llm-classifier-endpoint", default=os.getenv("MODELHUB_LLM_CLASSIFIER_ENDPOINT"), help=argparse.SUPPRESS) parser.add_argument("--llm-classifier-model", default=os.getenv("MODELHUB_LLM_CLASSIFIER_MODEL"), help=argparse.SUPPRESS) @@ -245,33 +239,11 @@ def _build_modelhub_client(base_args: argparse.Namespace) -> ModelHubClientPool: clients, capacity_probe_interval_cycles=max(0, int(getattr(base_args, "capacity_probe_interval_cycles", 3) or 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", 10) or 0)), + 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)), ) -def resolve_age_cleanup_policy( - base_args: argparse.Namespace, - *, - initial_cleanup_pending: bool, -) -> tuple[str, int]: - """Reserve ten slots initially, then use a five-slot cleanup hysteresis.""" - initial_reserve_slots = max( - 0, - int(getattr(base_args, "recent_model_reserve_slots", 10) or 0), - ) - dynamic_reserve_slots = min( - initial_reserve_slots, - max( - 0, - int(getattr(base_args, "dynamic_old_model_cleanup_reserve_slots", 5) or 0), - ), - ) - if initial_cleanup_pending: - return "initial", initial_reserve_slots - return "dynamic", dynamic_reserve_slots - - def _persist_architecture_blacklist( report: dict[str, Any], *, @@ -554,7 +526,6 @@ def run_poll_loop( submitted_total = 0 cycles = 0 stopped_reason = "max_cycles_reached" - initial_age_cleanup_pending = True pending_architecture_cleanup = False last_cleaned_architecture_blocks: set[str] = set() @@ -645,14 +616,9 @@ def run_poll_loop( ) if active_architecture_block_keys - last_cleaned_architecture_blocks: pending_architecture_cleanup = True - age_cleanup_mode, age_cleanup_reserve_slots = resolve_age_cleanup_policy( - base_args, - initial_cleanup_pending=initial_age_cleanup_pending, - ) log( - f"[queue-cleanup] mode={'architecture_dynamic' if architecture_only_cleanup else age_cleanup_mode} " - f"reserve_recent_slots={age_cleanup_reserve_slots} " - f"recent_days={max(1, int(getattr(base_args, 'recent_model_days', 7) or 7))}" + "[queue-cleanup] age_cleanup=disabled_admission_only " + f"deterministic_cleanup={'architecture_only' if architecture_only_cleanup else 'full'}" ) cleanup_summary = cleanup_certain_oom_tasks( modelhub_client, @@ -672,7 +638,6 @@ def run_poll_loop( ) ), architecture_only=architecture_only_cleanup, - age_reserved_slots=age_cleanup_reserve_slots, log=log, ) policy_cancelled_recorded = outcome_tracker.mark_policy_cancellations( @@ -697,9 +662,13 @@ def run_poll_loop( "mode": ( "architecture_dynamic" if architecture_only_cleanup - else age_cleanup_mode + else "deterministic_full" + ), + "ageCleanupMode": "admission_only", + "ageReservedSlots": max( + 0, + int(getattr(base_args, "recent_model_reserve_slots", 5) or 0), ), - "ageReservedSlots": age_cleanup_reserve_slots, "ageQueueThresholds": cleanup_summary["oldModelQueueThresholds"], "activeScanned": cleanup_summary["activeScanned"], "certainOomCount": cleanup_summary["certainOomCount"], @@ -715,8 +684,6 @@ def run_poll_loop( "stopErrorCount": len(cleanup_summary["stopErrors"]), } ) - if not architecture_only_cleanup: - initial_age_cleanup_pending = False pending_architecture_cleanup = False last_cleaned_architecture_blocks = active_architecture_block_keys except Exception as exc: diff --git a/modelhub_submmit_api/queue_cleanup.py b/modelhub_submmit_api/queue_cleanup.py index 82f523fa..bbc7c953 100644 --- a/modelhub_submmit_api/queue_cleanup.py +++ b/modelhub_submmit_api/queue_cleanup.py @@ -3,13 +3,12 @@ from __future__ import annotations import argparse from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass -from datetime import datetime, timedelta from pathlib import Path from typing import Any, Callable, Iterable from architecture_compatibility import architecture_compatibility_key, architecture_profiles from candidate_preflight import CandidatePreflightAdvisor, MODEL_LOAD_OVERHEAD -from common import utc_now, write_json +from common import write_json from hf_discovery import HuggingFaceDiscovery, inspect_repo_tree from modelhub_client import ModelHubClient, ModelHubClientPool from runner_common import DEFAULT_KEY_PATH, ensure_tokens @@ -168,41 +167,6 @@ def _load_repository_sizes( return sizes, errors -def _load_model_last_modified( - model_ids: set[str], - *, - discovery: HuggingFaceDiscovery, - read_concurrency: int, - log: Callable[[str], None], -) -> tuple[dict[str, datetime], dict[str, str]]: - values: dict[str, datetime] = {} - errors: dict[str, str] = {} - completed = 0 - workers = min(max(1, int(read_concurrency)), max(1, len(model_ids))) - with ThreadPoolExecutor(max_workers=workers) as executor: - futures = { - executor.submit(discovery.get_model_last_modified, model_id): model_id - for model_id in sorted(model_ids) - } - for future in as_completed(futures): - model_id = futures[future] - try: - value = future.result() - if value is None: - errors[model_id] = "model_last_modified_unknown" - else: - values[model_id] = value - except Exception as exc: - errors[model_id] = f"{type(exc).__name__}: {exc}" - completed += 1 - if completed == len(model_ids) or completed % 50 == 0: - log( - f"[queue-cleanup] age_scan={completed}/{len(model_ids)} " - f"complete={len(values)} unknown={len(errors)}" - ) - return values, errors - - def _load_model_configs( model_ids: set[str], *, @@ -480,74 +444,6 @@ def find_architecture_incompatible_tasks( return decisions, skipped -def find_old_overflow_tasks( - tasks: list[OwnedTask], - *, - model_last_modified: dict[str, datetime], - queue_threshold: int | dict[int, int] = 80, - recent_model_days: int = 7, - reference_time: datetime | None = None, - incomplete_accounts: set[int] | None = None, -) -> tuple[list[dict[str, Any]], dict[str, int]]: - recent_days = max(1, int(recent_model_days)) - cutoff = (reference_time or utc_now()) - timedelta(days=recent_days) - incomplete_accounts = incomplete_accounts or set() - grouped: dict[int, list[OwnedTask]] = {} - for task in tasks: - grouped.setdefault(task.account_index, []).append(task) - - decisions: list[dict[str, Any]] = [] - skipped = { - "accountsWithIncompleteListing": len(incomplete_accounts), - "withinFirstQueuePositions": 0, - "recentOverflowTasks": 0, - "modelAgeUnknown": 0, - "accountThresholdUnknown": 0, - "runningOverflowProtected": 0, - } - for account_index, account_tasks in sorted(grouped.items()): - if account_index in incomplete_accounts: - continue - if isinstance(queue_threshold, dict): - configured_threshold = queue_threshold.get(account_index) - if configured_threshold is None: - skipped["accountThresholdUnknown"] += 1 - continue - else: - configured_threshold = queue_threshold - threshold = max(0, int(configured_threshold)) - ordered = sorted(account_tasks, key=lambda item: item.task_id) - skipped["withinFirstQueuePositions"] += min(threshold, len(ordered)) - for position, task in enumerate(ordered, start=1): - if position <= threshold: - continue - if task.status != "waiting": - skipped["runningOverflowProtected"] += 1 - continue - last_modified = model_last_modified.get(task.model_id) - if last_modified is None: - skipped["modelAgeUnknown"] += 1 - continue - if last_modified >= cutoff: - skipped["recentOverflowTasks"] += 1 - continue - decisions.append( - { - "accountIndex": account_index + 1, - "taskId": task.task_id, - "modelId": task.model_id, - "gpuType": task.gpu_type, - "status": task.status, - "queuePosition": position, - "queueThreshold": threshold, - "modelLastModified": last_modified.isoformat(), - "recentCutoff": cutoff.isoformat(), - "reason": "old_model_beyond_account_queue_threshold", - } - ) - return decisions, skipped - - def _chunks(values: list[int], size: int) -> Iterable[list[int]]: for offset in range(0, len(values), size): yield values[offset : offset + size] @@ -564,18 +460,11 @@ def cleanup_certain_oom_tasks( architecture_compatibility_blocks: dict[str, dict[str, Any]] | None = None, task_compatibility_contexts: dict[str, dict[str, Any]] | None = None, architecture_only: bool = False, - age_reserved_slots: int | None = None, - reference_time: datetime | None = None, log: Callable[[str], None] = print, ) -> dict[str, Any]: - """Stop deterministic OOM/architecture tasks and over-threshold old tasks.""" + """Stop only deterministically impossible OOM/architecture tasks.""" clients = list(modelhub.clients) - reference_time = reference_time or utc_now() - configured_reserved_slots = ( - age_reserved_slots - if age_reserved_slots is not None - else getattr(modelhub, "recent_model_reserve_slots", 10) - ) + configured_reserved_slots = getattr(modelhub, "recent_model_reserve_slots", 5) reserved_slots = max(0, int(configured_reserved_slots or 0)) recent_model_days = max(1, int(getattr(modelhub, "recent_model_days", 7) or 7)) tasks, listing_errors = collect_active_tasks(clients, read_concurrency=read_concurrency) @@ -604,6 +493,11 @@ def cleanup_certain_oom_tasks( index: max(0, int(account_caps[index]) - reserved_slots) for index in range(min(len(clients), len(account_caps))) } + log( + f"[queue-cleanup] age_cleanup=disabled deterministic_cleanup=enabled " + f"age_policy=admission_only reserve_recent_slots={reserved_slots} " + f"recent_days={recent_model_days}" + ) model_ids = {task.model_id for task in tasks} repository_sizes: dict[str, int] = {} @@ -717,62 +611,6 @@ def cleanup_certain_oom_tasks( f"running_protected={architecture_skipped['runningMatchedProtected']}" ) - deterministic_task_keys = { - (int(decision["accountIndex"]) - 1, int(decision["taskId"])) - for decision in [*oom_decisions, *architecture_decisions] - } - # Deterministically impossible tasks are stopped first. Rank the age-policy - # queue as it will look afterwards to avoid over-cancelling old models. - age_rank_tasks = [ - task - for task in tasks - if (task.account_index, task.task_id) not in deterministic_task_keys - ] - overflow_model_ids = { - task.model_id - for account_index in range(len(clients)) - if account_index not in listing_errors - for task in sorted( - (item for item in age_rank_tasks if item.account_index == account_index), - key=lambda item: item.task_id, - )[queue_thresholds.get(account_index, len(age_rank_tasks)):] - if task.status == "waiting" - } - model_last_modified: dict[str, datetime] = {} - age_errors: dict[str, str] = {} - old_overflow_decisions: list[dict[str, Any]] = [] - age_skipped = { - "accountsWithIncompleteListing": len(listing_errors), - "withinFirstQueuePositions": 0, - "recentOverflowTasks": 0, - "modelAgeUnknown": 0, - "accountThresholdUnknown": 0, - "runningOverflowProtected": 0, - } - if not architecture_only: - model_last_modified, age_errors = _load_model_last_modified( - overflow_model_ids, - discovery=discovery, - read_concurrency=read_concurrency, - log=log, - ) - old_overflow_decisions, age_skipped = find_old_overflow_tasks( - age_rank_tasks, - model_last_modified=model_last_modified, - queue_threshold=queue_thresholds, - recent_model_days=recent_model_days, - reference_time=reference_time, - incomplete_accounts=set(listing_errors), - ) - log( - f"[queue-cleanup] old_overflow={len(old_overflow_decisions)} " - f"thresholds={','.join(str(queue_thresholds[index]) for index in sorted(queue_thresholds))} " - f"reserve_recent_slots={reserved_slots} recent_days={recent_model_days} " - f"recent_overflow={age_skipped['recentOverflowTasks']} " - f"age_unknown={age_skipped['modelAgeUnknown']} " - f"running_protected={age_skipped['runningOverflowProtected']}" - ) - decisions_by_key: dict[tuple[int, int], dict[str, Any]] = {} for decision in [*oom_decisions, *architecture_decisions]: enriched = dict(decision) @@ -790,17 +628,6 @@ def cleanup_certain_oom_tasks( if field not in {"reason", "cleanupReasons"} and value is not None } ) - for decision in old_overflow_decisions: - key = (int(decision["accountIndex"]), int(decision["taskId"])) - existing = decisions_by_key.get(key) - if existing is None: - enriched = dict(decision) - enriched["cleanupReasons"] = [decision["reason"]] - decisions_by_key[key] = enriched - continue - existing["cleanupReasons"].append(decision["reason"]) - for field in ("queuePosition", "queueThreshold", "modelLastModified", "recentCutoff"): - existing[field] = decision[field] decisions = sorted( decisions_by_key.values(), key=lambda item: (int(item["accountIndex"]), int(item["taskId"])), @@ -815,26 +642,10 @@ def cleanup_certain_oom_tasks( # query fails for that account, fail closed and do not terminate its tasks. refreshed_tasks, refresh_errors = collect_active_tasks(clients, read_concurrency=read_concurrency) active_ids_by_account: dict[int, set[int]] = {} - active_positions_by_account: dict[int, dict[int, int]] = {} active_status_by_account: dict[int, dict[int, str]] = {} for task in refreshed_tasks: active_ids_by_account.setdefault(task.account_index, set()).add(task.task_id) active_status_by_account.setdefault(task.account_index, {})[task.task_id] = task.status - for account_index in range(len(clients)): - planned_deterministic_ids = { - int(decision["taskId"]) - for decision in [*oom_decisions, *architecture_decisions] - if int(decision["accountIndex"]) - 1 == account_index - } - ordered_ids = sorted( - task.task_id - for task in refreshed_tasks - if task.account_index == account_index - and task.task_id not in planned_deterministic_ids - ) - active_positions_by_account[account_index] = { - task_id: position for position, task_id in enumerate(ordered_ids, start=1) - } for account_index, details in sorted(refresh_errors.items()): stop_errors.append( @@ -854,45 +665,20 @@ def cleanup_certain_oom_tasks( disappeared.append(decision) continue cleanup_reasons = set(decision.get("cleanupReasons") or [decision.get("reason")]) - age_only = cleanup_reasons == {"old_model_beyond_account_queue_threshold"} architecture_without_oom = bool( "known_framework_architecture_incompatible" in cleanup_reasons and "certain_oom_repository_size_exceeds_gpu_capacity" not in cleanup_reasons ) - current_position = active_positions_by_account.get(account_index, {}).get(int(decision["taskId"])) current_status = active_status_by_account.get(account_index, {}).get(int(decision["taskId"])) - account_queue_threshold = queue_thresholds.get(account_index) - if age_only and ( - current_position is None - or account_queue_threshold is None - or current_position <= account_queue_threshold - or current_status != "waiting" - ): - policy_no_longer_applies.append( - { - **decision, - "recheckedQueuePosition": current_position, - "recheckedStatus": current_status, - "policyChangeReason": ( - "task_started_running" - if current_status == "running" - else "queue_position_or_status_changed" - ), - } - ) - continue if architecture_without_oom and current_status != "waiting": policy_no_longer_applies.append( { **decision, - "recheckedQueuePosition": current_position, "recheckedStatus": current_status, "policyChangeReason": "task_started_running", } ) continue - if current_position is not None: - decision["recheckedQueuePosition"] = current_position by_account.setdefault(account_index, []).append(decision) batch_size = max(1, min(100, int(stop_batch_size))) @@ -901,7 +687,7 @@ def cleanup_certain_oom_tasks( if stop_failed: break decisions_by_id = {int(item["taskId"]): item for item in by_account[account_index]} - for cleanup_phase in ("oom", "architecture", "age"): + for cleanup_phase in ("oom", "architecture"): task_ids = sorted( task_id for task_id, decision in decisions_by_id.items() @@ -917,20 +703,10 @@ def cleanup_certain_oom_tasks( and "known_framework_architecture_incompatible" in decision["cleanupReasons"] ) - or ( - cleanup_phase == "age" - and "certain_oom_repository_size_exceeds_gpu_capacity" - not in decision["cleanupReasons"] - and "known_framework_architecture_incompatible" - not in decision["cleanupReasons"] - and "old_model_beyond_account_queue_threshold" - in decision["cleanupReasons"] - ) ) if cleanup_phase != "oom" and task_ids: - # Earlier deterministic stops can change positions, and a - # waiting architecture/age task can start running after the - # account-wide recheck. Re-read before each later phase. + # A waiting architecture task can start running after the + # account-wide recheck. Re-read before this protected phase. try: phase_tasks: dict[int, OwnedTask] = {} for status in ACTIVE_FILTER_STATUSES: @@ -951,18 +727,12 @@ def cleanup_certain_oom_tasks( stop_failed = True break - phase_positions = { - task_id: position - for position, task_id in enumerate(sorted(phase_tasks), start=1) - } eligible_task_ids: list[int] = [] - account_queue_threshold = queue_thresholds.get(account_index) for task_id in task_ids: current_task = phase_tasks.get(task_id) if current_task is None: disappeared.append(decisions_by_id[task_id]) continue - current_position = phase_positions.get(task_id) reasons = set( decisions_by_id[task_id].get("cleanupReasons") or [decisions_by_id[task_id].get("reason")] @@ -972,29 +742,19 @@ def cleanup_certain_oom_tasks( and "known_framework_architecture_incompatible" in reasons and current_task.status == "waiting" ) - age_applies = bool( - cleanup_phase == "age" - and "old_model_beyond_account_queue_threshold" in reasons - and account_queue_threshold is not None - and current_position is not None - and current_position > account_queue_threshold - and current_task.status == "waiting" - ) - if not architecture_applies and not age_applies: + if not architecture_applies: policy_no_longer_applies.append( { **decisions_by_id[task_id], - "recheckedQueuePosition": current_position, "recheckedStatus": current_task.status, "policyChangeReason": ( "task_started_running" if current_task.status == "running" - else "queue_position_status_or_policy_changed" + else "task_no_longer_architecture_cleanup_eligible" ), } ) continue - decisions_by_id[task_id]["recheckedQueuePosition"] = current_position eligible_task_ids.append(task_id) task_ids = eligible_task_ids @@ -1030,8 +790,9 @@ def cleanup_certain_oom_tasks( "uniqueModels": len(model_ids), "repositorySizesComplete": len(repository_sizes), "repositorySizeErrors": size_errors, - "modelAgeMetadataComplete": len(model_last_modified), - "modelAgeErrors": age_errors, + "ageCleanupMode": "admission_only", + "modelAgeMetadataComplete": 0, + "modelAgeErrors": {}, "listingErrors": {str(index + 1): values for index, values in listing_errors.items()}, "certainOomCount": len(oom_decisions), "certainOomTasks": oom_decisions, @@ -1046,8 +807,8 @@ def cleanup_certain_oom_tasks( }, "architectureFrameworkCatalogErrors": framework_catalog_errors, "architecturePolicySkipped": architecture_skipped, - "oldOverflowCount": len(old_overflow_decisions), - "oldOverflowTasks": old_overflow_decisions, + "oldOverflowCount": 0, + "oldOverflowTasks": [], "oldModelQueueThresholds": [ queue_thresholds.get(index) for index in range(len(clients)) ], @@ -1057,7 +818,10 @@ def cleanup_certain_oom_tasks( ], "recentModelReserveSlots": reserved_slots, "recentModelDays": recent_model_days, - "agePolicySkipped": age_skipped, + "agePolicySkipped": { + "cleanupDisabled": True, + "reason": "admission_only", + }, "cleanupCandidateCount": len(decisions), "skipped": skipped, "cancelledCount": len(cancelled), @@ -1071,7 +835,7 @@ def cleanup_certain_oom_tasks( def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( - description="Safely stop deterministic OOM/architecture and over-threshold old ModelHub tasks." + description="Safely stop only deterministic OOM/architecture ModelHub tasks." ) parser.add_argument("--execute", action="store_true", help="Actually terminate selected tasks; otherwise only print a preview") parser.add_argument("--read-concurrency", type=int, default=6) @@ -1105,7 +869,7 @@ def main(argv: list[str] | None = None) -> int: print( f"[queue-cleanup] finished certain_oom={summary['certainOomCount']} " f"architecture_incompatible={summary['architectureIncompatibleCount']} " - f"old_overflow={summary['oldOverflowCount']} " + f"age_cleanup={summary['ageCleanupMode']} " f"cancelled={summary['cancelledCount']} stop_errors={len(summary['stopErrors'])} " f"report={report_path}", flush=True, diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index eacaee6c..3bb782ab 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.12.5" +AGENT_VERSION = "2026.08.12.6" diff --git a/tests/test_candidate_preflight.py b/tests/test_candidate_preflight.py index 8336245b..7a27d34b 100644 --- a/tests/test_candidate_preflight.py +++ b/tests/test_candidate_preflight.py @@ -385,6 +385,8 @@ class CandidatePreflightTests(unittest.TestCase): model_profile={"modelType": "qwen3", "quantizationMethod": "awq"}, ) tracker._records[0]["outcome"] = "failed" # noqa: SLF001 + tracker._records[0]["failureCategory"] = "model_load" # noqa: SLF001 + tracker._records[0]["failureScope"] = "model_gpu_framework" # noqa: SLF001 report = tracker.get_stats_report() key = "gpu|vllm|text-generation|qwen3|awq" @@ -830,8 +832,50 @@ class CandidatePreflightTests(unittest.TestCase): self.assertEqual(1, combo["platformFailureCount"]) self.assertEqual(0, profile["consecutiveFailures"]) self.assertFalse(tracker.is_model_gpu_failed("owner/model", "gpu")) + self.assertEqual([], tracker.get_strategy_history_records()) self.assertNotIn("logCosUrl", read_jsonl(path)[0]) + def test_ambiguous_failure_is_unresolved_and_does_not_poison_strategy_feedback(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + path = Path(temporary_dir) / "outcomes.jsonl" + tracker = OutcomeTracker(path) + tracker.record_submission( + "owner/model", + "gpu", + "vllm", + "text-generation", + "task-ambiguous", + datetime.now(timezone.utc).isoformat(), + ) + task = { + "taskId": "task-ambiguous", + "status": "failed", + "verifyResult": -1, + "logCosUrl": "https://logs.invalid/task-ambiguous.zip", + } + classification = { + "failureCategory": "ambiguous_runtime", + "failureScope": "unknown", + "failureAction": "offline_review", + "failureNeedsLlm": False, + } + with patch( + "outcome_tracker.fetch_and_classify_failure_log", + return_value=classification, + ): + tracker.sync_from_api(TaskClient([task])) # type: ignore[arg-type] + + combo = tracker.get_stats_report()["combinationStats"][ + "gpu|vllm|text-generation" + ] + + self.assertEqual(1, combo["failureCount"]) + self.assertEqual(0, combo["attributableFailureCount"]) + self.assertEqual(0, combo["platformFailureCount"]) + self.assertEqual(1, combo["unresolvedFailureCount"]) + self.assertFalse(tracker.is_model_gpu_failed("owner/model", "gpu")) + self.assertEqual([], tracker.get_strategy_history_records()) + def test_failed_log_enrichment_attempt_is_persisted_and_bounded(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: path = Path(temporary_dir) / "outcomes.jsonl" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index c8619756..041bd4a2 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -215,21 +215,21 @@ def make_candidate(index: int) -> dict: class ClientPoolConcurrencyTests(unittest.TestCase): - def test_old_models_reserve_each_accounts_last_ten_queue_positions(self) -> None: - below_threshold = FakeClient(active_count=89) - at_threshold = FakeClient(active_count=90) + 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) pool = ModelHubClientPool( [below_threshold, at_threshold], # type: ignore[list-item] active_task_cap=100, active_counts_ttl=60, - recent_model_reserve_slots=10, + recent_model_reserve_slots=5, recent_model_days=7, instance_id="old-model-threshold-test", ) submitted_at = datetime(2026, 8, 11, tzinfo=timezone.utc) pool.add_task_for_model( - {"model": "old-allowed-as-position-90"}, + {"model": "old-allowed-as-position-95"}, model_last_modified=datetime(2026, 7, 1, tzinfo=timezone.utc), submitted_at=submitted_at, ) @@ -251,13 +251,13 @@ class ClientPoolConcurrencyTests(unittest.TestCase): ) self.assertEqual(2, len(below_threshold.submitted) + len(at_threshold.submitted)) - def test_concurrent_old_model_submissions_cannot_enter_reserved_ten_slots(self) -> None: - client = FakeClient(active_count=88) + def test_concurrent_old_model_submissions_cannot_enter_reserved_five_slots(self) -> None: + client = FakeClient(active_count=93) pool = ModelHubClientPool( [client], # type: ignore[list-item] active_task_cap=100, active_counts_ttl=60, - recent_model_reserve_slots=10, + recent_model_reserve_slots=5, recent_model_days=7, ) submitted_at = datetime(2026, 8, 11, tzinfo=timezone.utc) @@ -295,11 +295,11 @@ class ClientPoolConcurrencyTests(unittest.TestCase): self.assertTrue(all(stage["updatedAfter"] >= now - timedelta(days=7) for stage in stages)) def test_submit_candidate_reports_old_model_policy_skip_at_dynamic_threshold(self) -> None: - client = FakeClient(active_count=90) + client = FakeClient(active_count=95) pool = ModelHubClientPool( [client], # type: ignore[list-item] active_task_cap=100, - recent_model_reserve_slots=10, + recent_model_reserve_slots=5, recent_model_days=7, ) result = submit_candidate( @@ -315,25 +315,118 @@ class ClientPoolConcurrencyTests(unittest.TestCase): pool, # type: ignore[arg-type] ) - self.assertEqual("old_model_policy_skipped", result["outcome"]) + self.assertEqual("age_policy_deferred", result["outcome"]) + self.assertEqual("age_policy_skipped", result["reason"]) self.assertEqual([], client.submitted) - def test_old_model_threshold_tracks_a_discovered_capacity_increase(self) -> None: - client = DynamicCapacityClient(active=100, limit=101) + def test_age_policy_deferred_candidate_is_skipped_not_failed(self) -> None: + class AdmissionRaceClient: + @staticmethod + def begin_cycle() -> None: + return None + + @staticmethod + def old_model_submit_slots() -> int: + return 1 + + @staticmethod + def old_model_queue_thresholds() -> list[int]: + return [95] + + @staticmethod + def available_submit_slots() -> int: + return 1 + + @staticmethod + def processed_gpus_for_model(_model_id: str) -> set[str]: + return set() + + @staticmethod + def list_tasks_page(**_kwargs) -> dict: # noqa: ANN003 + return {"code": 0, "data": {"records": [], "pages": 0}} + + @staticmethod + def add_task_for_model(*_args, **_kwargs) -> dict: # noqa: ANN002, ANN003 + raise OldModelQueuePolicyError("old-model capacity filled during submission") + + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + args = build_parser().parse_args( + [ + "--gpus", + "Iluvatar_bi-150", + "--task-types", + "text-generation", + "--limit", + "1", + "--max-scan-models", + "1", + "--skip-outcome-sync", + "--skip-history-archive", + "--disable-candidate-preflight", + "--disable-gpu-strategy", + "--disable-market-intelligence", + ] + ) + args.runs_dir = str(root / "runs") + args.ledger_path = str(root / "ledger.jsonl") + args.outcomes_path = str(root / "outcomes.jsonl") + args.claims_path = str(root / "claims.jsonl") + args.history_archive_path = str(root / "history.jsonl") + args.submission_exclusions_path = str(root / "exclusions.jsonl") + client = AdmissionRaceClient() + + summary = run_submission( + args, + now=datetime(2026, 2, 1, tzinfo=timezone.utc), + hf_discovery=FakeDiscovery(1), # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + template_selector=TemplateSelector(), + ) + + self.assertEqual(0, summary["submittedCount"]) + self.assertEqual(0, summary["failedCount"]) + self.assertEqual(1, summary["skipReasonCounts"]["age_policy_skipped"]) + + def test_old_model_threshold_tracks_capacity_increase_from_100_to_200(self) -> None: + client = DynamicCapacityClient(active=100, limit=200) pool = ModelHubClientPool( [client], # type: ignore[list-item] active_task_cap=100, - recent_model_reserve_slots=10, + recent_model_reserve_slots=5, capacity_probe_interval_cycles=1, capacity_state_path=None, ) - self.assertEqual([90], pool.old_model_queue_thresholds()) - pool.configure_capacity_probe(1) - pool.add_task({"model": "recent-capacity-probe"}) + self.assertEqual([100], pool.active_task_counts()) + self.assertEqual([95], pool.old_model_queue_thresholds()) + pool.observe_capacity_lower_bounds([200]) - self.assertEqual([101], pool.account_capacity_limits()) - self.assertEqual([91], pool.old_model_queue_thresholds()) + self.assertEqual([200], pool.account_capacity_limits()) + self.assertEqual([195], pool.old_model_queue_thresholds()) + + def test_old_model_admission_fails_closed_when_active_count_is_unknown(self) -> None: + class UnreadableCountClient(FakeClient): + def count_active_tasks(self, **_kwargs) -> int: # noqa: ANN003 + raise TimeoutError("count unavailable") + + client = UnreadableCountClient(active_count=0) + pool = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + recent_model_reserve_slots=5, + capacity_state_path=None, + ) + + with self.assertRaises(OldModelQueuePolicyError): + pool.add_task_for_model( + {"model": "old-deferred"}, + model_last_modified=datetime(2026, 7, 1, tzinfo=timezone.utc), + submitted_at=datetime(2026, 8, 11, tzinfo=timezone.utc), + ) + + self.assertEqual([None], pool.old_model_queue_thresholds()) + self.assertEqual([], client.submitted) def test_online_submission_does_not_construct_llm_even_when_key_is_present(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: @@ -617,6 +710,8 @@ class ProcessCoordinationTests(unittest.TestCase): "modelId": "owner/old-failure", "targetGpu": "gpu-a", "outcome": "failed", + "failureCategory": "model_load", + "failureScope": "model_gpu_framework", "submitTime": (now - timedelta(hours=26)).isoformat(), "lastSyncTime": (now - timedelta(hours=25)).isoformat(), }, @@ -624,6 +719,8 @@ class ProcessCoordinationTests(unittest.TestCase): "modelId": "owner/recent-failure", "targetGpu": "gpu-a", "outcome": "failed", + "failureCategory": "model_load", + "failureScope": "model_gpu_framework", "submitTime": (now - timedelta(hours=2)).isoformat(), "lastSyncTime": (now - timedelta(hours=1)).isoformat(), }, diff --git a/tests/test_gpu_strategy.py b/tests/test_gpu_strategy.py index 05d61f03..8960e455 100644 --- a/tests/test_gpu_strategy.py +++ b/tests/test_gpu_strategy.py @@ -133,6 +133,34 @@ class GPUStrategyTests(unittest.TestCase): self.assertEqual(200, manager.state["acceptedTotal"]) self.assertEqual(0, manager.state["acceptedSinceRefresh"]) + def test_refresh_prefers_classified_attributable_outcomes(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + path = Path(temporary_dir) / "strategy.json" + raw_history = make_history( + "gpu-a", + 0, + 200, + start=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + classified_history = make_history( + "gpu-b", + 200, + 0, + start=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + client = HistoryClient(raw_history) + manager = GPUStrategyManager(path, long_term_min_samples=100) + + state = manager.prepare( + client, + supported_gpus=["gpu-a", "gpu-b", "gpu-c"], + history_records=classified_history, + ) + + self.assertEqual(0, client.list_calls) + self.assertEqual("classified_attributable_outcomes", state["historySource"]) + self.assertEqual("gpu-b", state["longTermGpus"][0]) + def test_market_weights_change_gpu_mix_without_changing_70_30_split(self) -> None: supported = ["gpu-a", "gpu-b", "gpu-c", "gpu-d"] manager = GPUStrategyManager("unused.json", market_intelligence=WeightedMarket()) diff --git a/tests/test_poll_policy.py b/tests/test_poll_policy.py index 9d779f10..57c82ad9 100644 --- a/tests/test_poll_policy.py +++ b/tests/test_poll_policy.py @@ -1,6 +1,5 @@ from __future__ import annotations -import argparse import json import sys import tempfile @@ -18,7 +17,7 @@ from outcome_tracker import OutcomeTracker # noqa: E402 from poll_runner import ( # noqa: E402 _bootstrap_architecture_history, _load_task_compatibility_contexts, - resolve_age_cleanup_policy, + build_parser, ) @@ -107,31 +106,11 @@ class PollPolicyTests(unittest.TestCase): self.assertEqual("mindie", contexts["task-old"]["framework"]) self.assertEqual({}, contexts["task-old"]["modelProfile"]) - def test_age_cleanup_uses_minus_ten_once_then_minus_five(self) -> None: - args = argparse.Namespace( - recent_model_reserve_slots=10, - dynamic_old_model_cleanup_reserve_slots=5, - ) + def test_age_policy_defaults_to_admission_only_reserve_five(self) -> None: + args = build_parser().parse_args([]) - self.assertEqual( - ("initial", 10), - resolve_age_cleanup_policy(args, initial_cleanup_pending=True), - ) - self.assertEqual( - ("dynamic", 5), - resolve_age_cleanup_policy(args, initial_cleanup_pending=False), - ) - - def test_dynamic_cleanup_cannot_be_stricter_than_initial_cleanup(self) -> None: - args = argparse.Namespace( - recent_model_reserve_slots=10, - dynamic_old_model_cleanup_reserve_slots=20, - ) - - self.assertEqual( - ("dynamic", 10), - resolve_age_cleanup_policy(args, initial_cleanup_pending=False), - ) + self.assertEqual(5, args.recent_model_reserve_slots) + self.assertFalse(hasattr(args, "dynamic_old_model_cleanup_reserve_slots")) if __name__ == "__main__": diff --git a/tests/test_queue_cleanup.py b/tests/test_queue_cleanup.py index d598874a..9219c5d9 100644 --- a/tests/test_queue_cleanup.py +++ b/tests/test_queue_cleanup.py @@ -19,7 +19,6 @@ from queue_cleanup import ( # noqa: E402 cleanup_certain_oom_tasks, find_architecture_incompatible_tasks, find_certain_oom_tasks, - find_old_overflow_tasks, ) @@ -394,98 +393,34 @@ class QueueCleanupTests(unittest.TestCase): self.assertEqual("task_started_running", summary["policyNoLongerAppliesTasks"][0]["policyChangeReason"]) self.assertEqual([], client.stopped) - def test_old_models_use_each_accounts_own_capacity_minus_ten_threshold(self) -> None: - now = datetime(2026, 8, 11, tzinfo=timezone.utc) - tasks = [ - OwnedTask(0, index, "owner/old", "Iluvatar_bi-100", "waiting") - for index in range(1, 92) - ] - tasks.extend( - OwnedTask(1, 1000 + index, "owner/old", "Iluvatar_bi-100", "waiting") - for index in range(1, 192) - ) - tasks.append(OwnedTask(0, 92, "owner/recent", "Iluvatar_bi-100", "waiting")) - tasks.append(OwnedTask(0, 93, "owner/old", "Iluvatar_bi-100", "running")) - tasks.append(OwnedTask(1, 1192, "owner/recent", "Iluvatar_bi-100", "waiting")) - - selected, skipped = find_old_overflow_tasks( - tasks, - model_last_modified={ - "owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc), - "owner/recent": datetime(2026, 8, 10, tzinfo=timezone.utc), - }, - queue_threshold={0: 90, 1: 190}, - recent_model_days=7, - reference_time=now, - ) - - self.assertEqual([91, 1191], [item["taskId"] for item in selected]) - self.assertEqual([91, 191], [item["queuePosition"] for item in selected]) - self.assertEqual(2, skipped["recentOverflowTasks"]) - self.assertEqual(1, skipped["runningOverflowProtected"]) - - def test_age_cleanup_never_stops_running_overflow_task(self) -> None: + def test_initial_and_later_cleanup_never_stop_old_waiting_or_running_tasks(self) -> None: records = [ { "taskId": index, "modelId": "owner/old", "gpuType": "Iluvatar_bi-100", - "status": "running" if index == 91 else "waiting", + "status": "running" if index == 100 else "waiting", } - for index in range(1, 92) + for index in range(1, 101) ] client = FakeQueueClient(records) pool = ModelHubClientPool( [client], # type: ignore[list-item] active_task_cap=100, - recent_model_reserve_slots=10, + recent_model_reserve_slots=5, ) - summary = cleanup_certain_oom_tasks( - pool, - FakeDiscovery( - {"owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), - log=lambda _message: None, - ) - - self.assertEqual(0, summary["oldOverflowCount"]) - self.assertEqual(1, summary["agePolicySkipped"]["runningOverflowProtected"]) - self.assertEqual([], client.stopped) - - def test_age_cleanup_recheck_releases_task_that_started_running(self) -> None: - records = [ - { - "taskId": index, - "modelId": "owner/old", - "gpuType": "Iluvatar_bi-100", - "status": "waiting", - } - for index in range(1, 92) + summaries = [ + cleanup_certain_oom_tasks( + pool, + FakeDiscovery({"owner/old": 1 * GIB}), # type: ignore[arg-type] + log=lambda _message: None, + ) + for _ in range(2) ] - # Read 1 is discovery, read 2 is the account-wide mutation recheck, - # and read 3 is the final age-only recheck after the OOM phase. - client = FakeQueueClient(records, promote_on_waiting_read=3) - pool = ModelHubClientPool( - [client], # type: ignore[list-item] - active_task_cap=100, - recent_model_reserve_slots=10, - ) - summary = cleanup_certain_oom_tasks( - pool, - FakeDiscovery( - {"owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - read_concurrency=1, - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), - log=lambda _message: None, - ) - self.assertEqual(1, summary["oldOverflowCount"]) - self.assertEqual(0, summary["cancelledCount"]) - self.assertEqual("task_started_running", summary["policyNoLongerAppliesTasks"][0]["policyChangeReason"]) + self.assertTrue(all(item["ageCleanupMode"] == "admission_only" for item in summaries)) + self.assertTrue(all(item["oldOverflowCount"] == 0 for item in summaries)) + self.assertTrue(all(item["cancelledCount"] == 0 for item in summaries)) self.assertEqual([], client.stopped) def test_certain_oom_cleanup_can_still_stop_running_task(self) -> None: @@ -510,38 +445,7 @@ class QueueCleanupTests(unittest.TestCase): self.assertEqual(1, summary["cancelledCount"]) self.assertEqual([[1]], client.stopped) - def test_old_overflow_task_is_not_stopped_if_it_moves_inside_dynamic_threshold(self) -> None: - records = [ - { - "taskId": index, - "modelId": "owner/old", - "gpuType": "Iluvatar_bi-100", - "status": "waiting", - } - for index in range(1, 92) - ] - client = FakeQueueClient(records, drop_first_on_recheck=True) - pool = ModelHubClientPool( - [client], # type: ignore[list-item] - active_task_cap=100, - recent_model_reserve_slots=10, - ) - summary = cleanup_certain_oom_tasks( - pool, - FakeDiscovery( - {"owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), - log=lambda _message: None, - ) - - self.assertEqual(1, summary["oldOverflowCount"]) - self.assertEqual(0, summary["cancelledCount"]) - self.assertEqual(1, summary["policyNoLongerAppliesCount"]) - self.assertEqual([], client.stopped) - - def test_cleanup_promotes_capacity_from_complete_active_listing(self) -> None: + def test_capacity_decline_pauses_submissions_and_never_deletes_existing_tasks(self) -> None: records = [ { "taskId": index, @@ -555,114 +459,20 @@ class QueueCleanupTests(unittest.TestCase): pool = ModelHubClientPool( [client], # type: ignore[list-item] active_task_cap=100, - recent_model_reserve_slots=10, + recent_model_reserve_slots=5, capacity_state_path=None, ) summary = cleanup_certain_oom_tasks( pool, - FakeDiscovery( - {"owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), + FakeDiscovery({"owner/old": 1 * GIB}), # type: ignore[arg-type] log=lambda _message: None, ) self.assertEqual([150], summary["accountCapacityLimits"]) - self.assertEqual([140], summary["oldModelQueueThresholds"]) - self.assertEqual(list(range(141, 151)), [item["taskId"] for item in summary["oldOverflowTasks"]]) - - def test_initial_cleanup_stops_old_task_beyond_capacity_minus_ten(self) -> None: - records = [ - { - "taskId": index, - "modelId": "owner/old", - "gpuType": "Iluvatar_bi-100", - "status": "waiting", - } - for index in range(1, 92) - ] - client = FakeQueueClient(records) - pool = ModelHubClientPool( - [client], # type: ignore[list-item] - active_task_cap=100, - recent_model_reserve_slots=10, - ) - summary = cleanup_certain_oom_tasks( - pool, - FakeDiscovery( - {"owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), - log=lambda _message: None, - ) - - self.assertEqual(1, summary["oldOverflowCount"]) - self.assertEqual(1, summary["cancelledCount"]) - self.assertEqual([90], summary["oldModelQueueThresholds"]) - self.assertEqual([[91]], client.stopped) - - def test_scheduled_cleanup_uses_capacity_minus_five(self) -> None: - records = [ - { - "taskId": index, - "modelId": "owner/old", - "gpuType": "Iluvatar_bi-100", - "status": "waiting", - } - for index in range(1, 97) - ] - client = FakeQueueClient(records) - pool = ModelHubClientPool( - [client], # type: ignore[list-item] - active_task_cap=100, - recent_model_reserve_slots=10, - ) - summary = cleanup_certain_oom_tasks( - pool, - FakeDiscovery( - {"owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - age_reserved_slots=5, - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), - log=lambda _message: None, - ) - - self.assertEqual([95], summary["oldModelQueueThresholds"]) - self.assertEqual([96], [item["taskId"] for item in summary["oldOverflowTasks"]]) - self.assertEqual([[96]], client.stopped) - - def test_oom_is_removed_before_recalculating_old_overflow_positions(self) -> None: - records = [ - { - "taskId": index, - "modelId": "owner/large" if index == 1 else "owner/old", - "gpuType": "Iluvatar_bi-100", - "status": "waiting", - } - for index in range(1, 93) - ] - client = FakeQueueClient(records) - pool = ModelHubClientPool( - [client], # type: ignore[list-item] - active_task_cap=100, - recent_model_reserve_slots=10, - ) - summary = cleanup_certain_oom_tasks( - pool, - FakeDiscovery( - {"owner/large": 40 * GIB, "owner/old": 1 * GIB}, - {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, - ), # type: ignore[arg-type] - reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), - log=lambda _message: None, - ) - - self.assertEqual(1, summary["certainOomCount"]) - self.assertEqual([92], [item["taskId"] for item in summary["oldOverflowTasks"]]) - self.assertEqual([[1], [92]], client.stopped) + self.assertEqual([145], summary["oldModelQueueThresholds"]) + self.assertEqual(0, summary["oldOverflowCount"]) + self.assertEqual(0, pool.available_submit_slots()) + self.assertEqual([], client.stopped) def test_stop_tasks_uses_documented_put_endpoint_and_integer_ids(self) -> None: http = RecordingHttpClient()