From 7ec875563e930e005d3b2a674291da115c7d4330 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Wed, 12 Aug 2026 08:19:53 +0800 Subject: [PATCH] feat: dynamically clean incompatible architectures --- README.md | 28 +- modelhub_submmit_api/README.md | 11 +- .../architecture_compatibility.py | 40 +- modelhub_submmit_api/candidate_preflight.py | 37 +- modelhub_submmit_api/failure_log_inspector.py | 35 ++ modelhub_submmit_api/outcome_tracker.py | 107 ++++- modelhub_submmit_api/poll_runner.py | 171 +++++++- modelhub_submmit_api/queue_cleanup.py | 382 +++++++++++++++--- modelhub_submmit_api/version.py | 2 +- tests/test_candidate_preflight.py | 72 ++++ tests/test_poll_policy.py | 43 +- tests/test_queue_cleanup.py | 183 +++++++++ 12 files changed, 979 insertions(+), 132 deletions(-) diff --git a/README.md b/README.md index 2028db66..77d3c075 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,8 @@ Optional tuning: - `MODELHUB_QUEUE_CLEANUP_INTERVAL_CYCLES` default `120`; cleanup also runs once at startup - `MODELHUB_QUEUE_CLEANUP_READ_CONCURRENCY` default `6` - `MODELHUB_QUEUE_CLEANUP_REPORT_PATH` default `.modelhub_state/queue_cleanup_latest.json` +- `MODELHUB_ARCHITECTURE_BLACKLIST_PATH` default `.modelhub_state/architecture_compatibility_blacklist.json` +- `MODELHUB_ARCHITECTURE_BLOCK_TTL_DAYS` default `30` - `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` default `10` per account - `MODELHUB_DYNAMIC_OLD_MODEL_CLEANUP_RESERVE_SLOTS` default `5` per account - `MODELHUB_RECENT_MODEL_DAYS` default `7` @@ -115,13 +117,14 @@ says that the selected framework does not support the model or architecture, the runner learns an exact GPU + framework + task type + architecture block from the candidate repository's `config.json`. Repository names are never used as architecture evidence. Exact `architectures` values take priority and -`model_type` is used only when `architectures` is absent; missing metadata does -not create a block. Generic unsupported operators, attention backends, GPU types, +`model_type` is also retained when the runtime explicitly says Transformers does +not recognize that type; missing metadata does not create a block. Generic unsupported operators, attention backends, GPU types, quantization failures, and OOMs cannot enter this blacklist. A newer success for the same exact combination clears the block, and otherwise it expires after 30 days. Set `MODELHUB_ARCHITECTURE_BLOCK_TTL_DAYS` to a value from 1 to 365 to change that window. The stats report exposes `architectureCompatibilityBlocks` -and per-GPU/framework block counts. +and per-GPU/framework block counts. The live snapshot is written to +`.modelhub_state/architecture_compatibility_blacklist.json`. Before a candidate reaches the submit queue, failure-informed preflight checks the actual ModelScope repository structure and file sizes. Non-GGUF text @@ -138,7 +141,7 @@ capacities with `MODELHUB_GPU_MEMORY_GIB_JSON`, for example `{"New_gpu": 64}`. -At poller startup, the same deterministic memory gate is applied to existing +At poller startup, the same deterministic memory and learned architecture gates are applied to existing `waiting` and `running` tasks across every configured account. A task is stopped through `PUT /api/async/task/stop-create-contest-task` only when its own current recursive repository size, multiplied by ModelHub's observed `1.20` overhead, @@ -151,6 +154,16 @@ 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; +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 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 @@ -270,12 +283,15 @@ Version `2026.08.12.2` learns conservative, expiring GPU/framework/architecture compatibility blocks only from explicit ModelHub failure text, matches candidate `config.json` metadata instead of repository names, and lets newer success evidence clear stale blocks. +Version `2026.08.12.3` extracts unsupported `model_type`/`architectures` from the +platform's fixed failure wording, persists a dynamically growing blacklist, and +immediately removes exact-matching waiting tasks with two active-state checks. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag agent-v20 -git push origin agent-v20 +git tag agent-v21 +git push origin agent-v21 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 83ce46bf..7677ab0a 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 active tasks that are certain to exceed GPU memory +- `queue_cleanup.py`: fail-closed cleanup for certain OOM, architecture, and age 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 @@ -96,9 +96,13 @@ bash run_poll.sh --dry-run sub-20% rate over the latest 20 terminal tasks pauses it for 6 hours. - An explicit "framework does not support this model/architecture" failure learns a 30-day GPU + framework + task + architecture block. Architecture identity - comes from candidate `config.json` (`architectures`, with `model_type` only as - fallback), never from repository names. A newer success clears the block, and + comes from candidate `config.json` plus exact unsupported `model_type` or + `architectures` strings in the runtime log, never from repository names. A newer success clears the block, and generic unsupported backend/operator messages cannot create one. +- Blacklist additions are persisted and detected every three poll cycles. A new + rule immediately launches a lightweight architecture-only queue scan. Exact + matching waiting tasks are stopped after two state checks; running tasks and + tasks without local framework/task metadata are protected. - A strategy generation lasts exactly 200 platform-accepted submissions. Rejected API calls and duplicates do not advance it. The next cycle refreshes platform history before submitting again. - Strategy state is stored in `.modelhub_state/gpu_strategy.json`; a generation never recalculates @@ -227,6 +231,7 @@ Persistent local scheduler state is written under `.modelhub_state/`: - `account_capacity.json`: learned per-account active-task limits - `submission_exclusions.jsonl`: non-retryable model/GPU uniqueness rejections - `queue_cleanup_latest.json`: latest active-task sizing evidence and cancellation result +- `architecture_compatibility_blacklist.json`: current dynamic compatibility blocks and evidence ## Verification diff --git a/modelhub_submmit_api/architecture_compatibility.py b/modelhub_submmit_api/architecture_compatibility.py index 60ab13d0..20f82d46 100644 --- a/modelhub_submmit_api/architecture_compatibility.py +++ b/modelhub_submmit_api/architecture_compatibility.py @@ -15,23 +15,37 @@ def architecture_profile( architectures: Any, ) -> dict[str, Any] | None: """Build a stable, conservative architecture identity for feedback matching.""" + profiles = architecture_profiles(model_type, architectures) + return profiles[0] if profiles else None + + +def architecture_profiles( + model_type: Any, + architectures: Any, +) -> list[dict[str, Any]]: + """Return exact architecture identity first, followed by model-type fallback.""" normalized_architectures = _normalize_architectures(architectures) normalized_model_type = _normalize(model_type) + profiles: list[dict[str, Any]] = [] if normalized_architectures: - return { - "matchType": "architectures", - "signature": "architectures:" + ",".join(normalized_architectures), - "architectures": normalized_architectures, - "modelType": normalized_model_type or None, - } + profiles.append( + { + "matchType": "architectures", + "signature": "architectures:" + ",".join(normalized_architectures), + "architectures": normalized_architectures, + "modelType": normalized_model_type or None, + } + ) if normalized_model_type: - return { - "matchType": "model_type", - "signature": f"model_type:{normalized_model_type}", - "architectures": [], - "modelType": normalized_model_type, - } - return None + profiles.append( + { + "matchType": "model_type", + "signature": f"model_type:{normalized_model_type}", + "architectures": [], + "modelType": normalized_model_type, + } + ) + return profiles def architecture_compatibility_key( diff --git a/modelhub_submmit_api/candidate_preflight.py b/modelhub_submmit_api/candidate_preflight.py index d720bb88..d0e0f3e6 100644 --- a/modelhub_submmit_api/candidate_preflight.py +++ b/modelhub_submmit_api/candidate_preflight.py @@ -8,7 +8,7 @@ from dataclasses import dataclass from datetime import timedelta from typing import Any -from architecture_compatibility import architecture_compatibility_key, architecture_profile +from architecture_compatibility import architecture_compatibility_key, architecture_profiles from llm_classifier import LLMAssistedClassifier from models import ModelInspection from common import parse_datetime, utc_now @@ -422,24 +422,23 @@ class CandidatePreflightAdvisor: framework: str, task_type: str, ) -> dict[str, Any] | None: - profile = architecture_profile(inspection.model_type, inspection.architectures) - if profile is None: - return None - key = architecture_compatibility_key( - target_gpu, - framework, - task_type, - profile["signature"], - ) - if key is None: - return None - block = self._architecture_compatibility_blocks.get(key) - if not isinstance(block, dict): - return None - expires_at = parse_datetime(block.get("expiresAt")) - if expires_at is None or expires_at <= utc_now(): - return None - return dict(block) + for profile in architecture_profiles(inspection.model_type, inspection.architectures): + key = architecture_compatibility_key( + target_gpu, + framework, + task_type, + profile["signature"], + ) + if key is None: + continue + block = self._architecture_compatibility_blocks.get(key) + if not isinstance(block, dict): + continue + expires_at = parse_datetime(block.get("expiresAt")) + if expires_at is None or expires_at <= utc_now(): + continue + return dict(block) + return None def summary(self) -> dict[str, Any]: with self._lock: diff --git a/modelhub_submmit_api/failure_log_inspector.py b/modelhub_submmit_api/failure_log_inspector.py index 882fcb0a..af51151e 100644 --- a/modelhub_submmit_api/failure_log_inspector.py +++ b/modelhub_submmit_api/failure_log_inspector.py @@ -22,6 +22,16 @@ ERROR_LINE_PATTERN = re.compile( r"找不到空闲卡|不支持|暂不支持|不兼容|请换用|请更换)", re.IGNORECASE, ) +MODEL_TYPE_NOT_RECOGNIZED_PATTERN = re.compile( + r"model\s+type\s+[`'\"](?P[A-Za-z0-9_.-]+)[`'\"]\s+but\s+" + r"(?:Transformers\s+)?does\s+not\s+recognize\s+this\s+architecture", + re.IGNORECASE, +) +MODEL_ARCHITECTURES_NOT_SUPPORTED_PATTERN = re.compile( + r"Model\s+architectures?\s*(?P\[[^\]\n]{1,500}\])\s+" + r"(?:are|is)\s+not\s+supported\s+for\s+now", + re.IGNORECASE, +) def fetch_and_classify_failure_log( @@ -94,6 +104,13 @@ def classify_failure_archive( observed_memory_gib = _extract_observed_gpu_memory_gib(error_lines) if report_code == "PREFLIGHT_OOM" and observed_memory_gib is not None: result["failureObservedGpuMemoryGiB"] = observed_memory_gib + unsupported_architectures, unsupported_model_types = _extract_unsupported_architectures( + error_lines + ) + if unsupported_architectures: + result["failureUnsupportedArchitectures"] = unsupported_architectures + if unsupported_model_types: + result["failureUnsupportedModelTypes"] = unsupported_model_types if classification.needs_llm and llm_classifier is not None and llm_classifier.enabled: llm_decision = llm_classifier.classify_failure( task_context=dict(task_context or {}), @@ -146,3 +163,21 @@ def _extract_observed_gpu_memory_gib(error_lines: list[str]) -> float | None: if 0 < value <= 1024: return value return None + + +def _extract_unsupported_architectures( + error_lines: list[str], +) -> tuple[list[str], list[str]]: + architectures: set[str] = set() + model_types: set[str] = set() + for line in error_lines: + for match in MODEL_TYPE_NOT_RECOGNIZED_PATTERN.finditer(line): + value = match.group("model_type").strip() + if value: + model_types.add(value) + for match in MODEL_ARCHITECTURES_NOT_SUPPORTED_PATTERN.finditer(line): + for value in re.findall(r"['\"]([^'\"]+)['\"]", match.group("architectures")): + value = value.strip() + if value: + architectures.add(value) + return sorted(architectures, key=str.casefold), sorted(model_types, key=str.casefold) diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index 7a1e901a..156232a1 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -13,6 +13,7 @@ from architecture_compatibility import ( EXPLICIT_ARCHITECTURE_FAILURE_REASON, architecture_compatibility_key, architecture_profile, + architecture_profiles, ) from common import append_jsonl, parse_datetime, read_jsonl, update_jsonl, utc_now from failure_log_inspector import fetch_and_classify_failure_log @@ -53,6 +54,26 @@ class OutcomeTracker: def set_failure_llm_classifier(self, classifier: LLMAssistedClassifier | None) -> None: self._failure_llm_classifier = classifier + def get_task_compatibility_contexts(self) -> dict[str, dict[str, Any]]: + """Return locally known submit metadata needed for account queue cleanup.""" + contexts: dict[str, dict[str, Any]] = {} + for task_id, record in self._by_task_id.items(): + framework = str(record.get("framework") or "").strip() + task_type = str(record.get("taskType") or "").strip() + if not framework or not task_type: + continue + profile = record.get("modelProfile") + contexts[task_id] = { + "taskId": task_id, + "modelId": str(record.get("modelId") or ""), + "targetGpu": str(record.get("targetGpu") or ""), + "framework": framework, + "taskType": task_type, + "modelProfile": dict(profile) if isinstance(profile, dict) else {}, + "submitTime": record.get("submitTime"), + } + return contexts + def _rebuild_indexes(self) -> None: self._by_task_id.clear() self._by_model_gpu.clear() @@ -176,9 +197,10 @@ class OutcomeTracker: self._by_model_gpu[(model_id, target_gpu)].append(record) updated_count += 1 - # Retry a small bounded set of our own failed submissions. Historical - # tasks without the locally recorded framework/profile are intentionally - # excluded to avoid downloading thousands of old log archives at once. + # Retry a small bounded set of our own failed submissions. A stored + # framework is sufficient: fixed MODEL_NOT_SUPPORTED log text can yield + # an exact architecture/model_type even for older records that predate + # local modelProfile capture. candidate_ids = {id(record) for record in enrichment_candidates} for record in self._records: if len(enrichment_candidates) >= FAILURE_ENRICHMENT_LIMIT: @@ -189,7 +211,6 @@ class OutcomeTracker: record.get("outcome") == "failed" and record.get("logCosUrl") and record.get("framework") - and record.get("modelProfile") and not record.get("failureCategory") and int(record.get("failureEnrichmentAttempts") or 0) < FAILURE_ENRICHMENT_MAX_ATTEMPTS ): @@ -537,36 +558,37 @@ def _build_architecture_compatibility_blocks( cutoff = now - timedelta(days=max(1, int(ttl_days))) for record in records: - profile_data = record.get("modelProfile") - if not isinstance(profile_data, dict): - continue - profile = architecture_profile( - profile_data.get("modelType"), - profile_data.get("architectures"), - ) - if profile is None: - continue target_gpu = str(record.get("targetGpu") or "").strip() framework = str(record.get("framework") or "").strip() task_type = str(record.get("taskType") or "").strip() - key = architecture_compatibility_key( - target_gpu, - framework, - task_type, - profile["signature"], - ) event_time = ( parse_datetime(record.get("submitTime")) or parse_datetime(record.get("lastSyncTime")) ) - if key is None or event_time is None: + if not target_gpu or not framework or not task_type or event_time is None: continue if record.get("outcome") == "success": - successes[key].append((event_time, record)) + for profile in _stored_architecture_profiles(record, include_model_type=True): + key = architecture_compatibility_key( + target_gpu, + framework, + task_type, + profile["signature"], + ) + if key is not None: + successes[key].append((event_time, record)) continue if not _is_explicit_architecture_failure(record) or event_time < cutoff: continue - failures[key].append((event_time, record, profile)) + for profile in _failure_architecture_profiles(record): + key = architecture_compatibility_key( + target_gpu, + framework, + task_type, + profile["signature"], + ) + if key is not None: + failures[key].append((event_time, record, profile)) blocks: dict[str, dict[str, Any]] = {} for key, failure_events in failures.items(): @@ -606,6 +628,47 @@ def _build_architecture_compatibility_blocks( return blocks +def _stored_architecture_profiles( + record: dict[str, Any], + *, + include_model_type: bool, +) -> list[dict[str, Any]]: + profile_data = record.get("modelProfile") + if not isinstance(profile_data, dict): + return [] + if include_model_type: + return architecture_profiles( + profile_data.get("modelType"), + profile_data.get("architectures"), + ) + profile = architecture_profile( + profile_data.get("modelType"), + profile_data.get("architectures"), + ) + return [profile] if profile is not None else [] + + +def _failure_architecture_profiles(record: dict[str, Any]) -> list[dict[str, Any]]: + profiles: list[dict[str, Any]] = [] + unsupported_architectures = record.get("failureUnsupportedArchitectures") + if isinstance(unsupported_architectures, list) and unsupported_architectures: + profile = architecture_profile(None, unsupported_architectures) + if profile is not None: + profiles.append(profile) + unsupported_model_types = record.get("failureUnsupportedModelTypes") + if isinstance(unsupported_model_types, list): + for model_type in unsupported_model_types: + profile = architecture_profile(model_type, None) + if profile is not None: + profiles.append(profile) + if not profiles: + profiles.extend(_stored_architecture_profiles(record, include_model_type=False)) + deduped: dict[str, dict[str, Any]] = {} + for profile in profiles: + deduped[profile["signature"]] = profile + return list(deduped.values()) + + def _is_explicit_architecture_failure(record: dict[str, Any]) -> bool: return bool( record.get("outcome") == "failed" diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 1f479a27..491f8447 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -8,7 +8,7 @@ import time from pathlib import Path from typing import Any, Callable -from common import utc_now, write_json +from common import read_jsonl, utc_now, write_json from daily_runner import DEFAULT_DAILY_RUNS_DIR, log, run_daily_batches from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH from hf_discovery import HuggingFaceDiscovery @@ -31,6 +31,9 @@ from version import AGENT_VERSION DEFAULT_POLL_RUNS_DIR = Path("poll_runs") +DEFAULT_ARCHITECTURE_BLACKLIST_PATH = Path( + ".modelhub_state/architecture_compatibility_blacklist.json" +) def build_parser() -> argparse.ArgumentParser: @@ -204,6 +207,14 @@ def build_parser() -> argparse.ArgumentParser: default=os.getenv("MODELHUB_QUEUE_CLEANUP_REPORT_PATH", ".modelhub_state/queue_cleanup_latest.json"), help=argparse.SUPPRESS, ) + parser.add_argument( + "--architecture-blacklist-path", + default=os.getenv( + "MODELHUB_ARCHITECTURE_BLACKLIST_PATH", + str(DEFAULT_ARCHITECTURE_BLACKLIST_PATH), + ), + help=argparse.SUPPRESS, + ) return parser @@ -251,6 +262,53 @@ def resolve_age_cleanup_policy( return "dynamic", dynamic_reserve_slots +def _persist_architecture_blacklist( + report: dict[str, Any], + *, + path: Path, +) -> set[str]: + blocks = report.get("architectureCompatibilityBlocks") or {} + blocks = blocks if isinstance(blocks, dict) else {} + write_json( + path, + { + "generatedAt": report.get("generatedAt"), + "summary": report.get("architectureCompatibilitySummary") or {}, + "blocks": blocks, + }, + ) + return set(str(key) for key in blocks) + + +def _load_task_compatibility_contexts( + outcome_tracker: OutcomeTracker, + *, + ledger_path: Path, +) -> dict[str, dict[str, Any]]: + contexts = outcome_tracker.get_task_compatibility_contexts() + for record in read_jsonl(ledger_path): + task_id_value = record.get("taskId") + if task_id_value is None: + continue + task_id = str(task_id_value) + existing = contexts.get(task_id) + if existing is None: + contexts[task_id] = { + "taskId": task_id, + "modelId": str(record.get("modelId") or ""), + "targetGpu": str(record.get("targetGpu") or ""), + "framework": str(record.get("framework") or ""), + "taskType": str(record.get("taskType") or ""), + "modelProfile": {}, + "submitTime": record.get("submitTime"), + } + continue + for field in ("modelId", "targetGpu", "framework", "taskType", "submitTime"): + if not existing.get(field) and record.get(field): + existing[field] = record.get(field) + return contexts + + def run_poll_loop( *, base_args: argparse.Namespace, @@ -286,6 +344,8 @@ def run_poll_loop( cycles = 0 stopped_reason = "max_cycles_reached" initial_age_cleanup_pending = True + pending_architecture_cleanup = False + last_cleaned_architecture_blocks: set[str] = set() while True: if base_args.max_cycles and cycles >= base_args.max_cycles: @@ -296,22 +356,90 @@ def run_poll_loop( if hasattr(modelhub_client, "configure_capacity_probe"): modelhub_client.configure_capacity_probe(cycles) + outcome_synced_this_cycle = False + if ( + cycles % OUTCOME_SYNC_INTERVAL == 0 + and not getattr(base_args, "skip_outcome_sync", False) + ): + try: + synced = outcome_tracker.sync_from_api(modelhub_client) + outcome_synced_this_cycle = True + if synced > 0: + log(f"[poll] cycle={cycles} outcome_sync_updated={synced}") + sync_feedback = outcome_tracker.get_stats_report() + active_block_keys = _persist_architecture_blacklist( + sync_feedback, + path=Path( + getattr( + base_args, + "architecture_blacklist_path", + DEFAULT_ARCHITECTURE_BLACKLIST_PATH, + ) + ), + ) + new_block_keys = active_block_keys - last_cleaned_architecture_blocks + if ( + new_block_keys + and not bool(getattr(base_args, "disable_queue_cleanup", False)) + and isinstance(modelhub_client, ModelHubClientPool) + ): + pending_architecture_cleanup = True + log( + f"[queue-cleanup] dynamic_architecture_blocks_added={len(new_block_keys)} " + f"cleanup_next=immediate" + ) + except Exception as exc: + log(f"[poll] cycle={cycles} outcome_sync_error={exc}") + cleanup_interval = max(0, int(getattr(base_args, "queue_cleanup_interval_cycles", 120) or 0)) + scheduled_queue_cleanup = bool( + cycles == 1 or (cleanup_interval > 0 and cycles % cleanup_interval == 0) + ) + architecture_only_cleanup = bool( + pending_architecture_cleanup and not scheduled_queue_cleanup + ) should_cleanup_queue = ( not bool(getattr(base_args, "disable_queue_cleanup", False)) and isinstance(modelhub_client, ModelHubClientPool) - and (cycles == 1 or (cleanup_interval > 0 and cycles % cleanup_interval == 0)) + and ( + scheduled_queue_cleanup + or pending_architecture_cleanup + ) ) if should_cleanup_queue: try: + if ( + not outcome_synced_this_cycle + and not getattr(base_args, "skip_outcome_sync", False) + ): + synced_before_cleanup = outcome_tracker.sync_from_api(modelhub_client) + if synced_before_cleanup: + log( + f"[queue-cleanup] outcome_sync_updated={synced_before_cleanup}" + ) cleanup_feedback = outcome_tracker.get_stats_report() cleanup_gpu_memory = cleanup_feedback.get("observedGpuMemoryGiB") or {} + cleanup_architecture_blocks = ( + cleanup_feedback.get("architectureCompatibilityBlocks") or {} + ) + active_architecture_block_keys = _persist_architecture_blacklist( + cleanup_feedback, + path=Path( + getattr( + base_args, + "architecture_blacklist_path", + DEFAULT_ARCHITECTURE_BLACKLIST_PATH, + ) + ), + ) + 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={age_cleanup_mode} " + 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))}" ) @@ -321,6 +449,18 @@ def run_poll_loop( dry_run=bool(base_args.dry_run), read_concurrency=max(1, int(getattr(base_args, "queue_cleanup_read_concurrency", 6) or 6)), gpu_memory_gib=cleanup_gpu_memory if isinstance(cleanup_gpu_memory, dict) else None, + architecture_compatibility_blocks=( + cleanup_architecture_blocks + if isinstance(cleanup_architecture_blocks, dict) + else None + ), + task_compatibility_contexts=( + _load_task_compatibility_contexts( + outcome_tracker, + ledger_path=Path(base_args.ledger_path), + ) + ), + architecture_only=architecture_only_cleanup, age_reserved_slots=age_cleanup_reserve_slots, log=log, ) @@ -343,18 +483,31 @@ def run_poll_loop( queue_cleanup_runs.append( { "cycle": cycles, - "mode": age_cleanup_mode, + "mode": ( + "architecture_dynamic" + if architecture_only_cleanup + else age_cleanup_mode + ), "ageReservedSlots": age_cleanup_reserve_slots, "ageQueueThresholds": cleanup_summary["oldModelQueueThresholds"], "activeScanned": cleanup_summary["activeScanned"], "certainOomCount": cleanup_summary["certainOomCount"], + "architectureBlockCount": cleanup_summary[ + "architectureBlockCount" + ], + "architectureIncompatibleCount": cleanup_summary[ + "architectureIncompatibleCount" + ], "oldOverflowCount": cleanup_summary["oldOverflowCount"], "cancelledCount": cleanup_summary["cancelledCount"], "policyCancelledRecorded": policy_cancelled_recorded, "stopErrorCount": len(cleanup_summary["stopErrors"]), } ) - initial_age_cleanup_pending = False + 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: log(f"[queue-cleanup] error={type(exc).__name__}: {exc} continue_polling=true") @@ -410,14 +563,6 @@ def run_poll_loop( time.sleep(base_args.idle_interval_seconds) continue - if cycles % OUTCOME_SYNC_INTERVAL == 0: - try: - synced = outcome_tracker.sync_from_api(modelhub_client) - if synced > 0: - log(f"[poll] cycle={cycles} outcome_sync_updated={synced}") - except Exception as exc: - log(f"[poll] cycle={cycles} outcome_sync_error={exc}") - if cycles % STATS_PRINT_INTERVAL == 0: try: stats = outcome_tracker.get_stats_report() diff --git a/modelhub_submmit_api/queue_cleanup.py b/modelhub_submmit_api/queue_cleanup.py index efe027e2..89f34459 100644 --- a/modelhub_submmit_api/queue_cleanup.py +++ b/modelhub_submmit_api/queue_cleanup.py @@ -7,6 +7,7 @@ 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 hf_discovery import HuggingFaceDiscovery, inspect_repo_tree @@ -195,6 +196,43 @@ def _load_model_last_modified( return values, errors +def _load_model_configs( + model_ids: set[str], + *, + discovery: HuggingFaceDiscovery, + read_concurrency: int, + log: Callable[[str], None], +) -> tuple[dict[str, dict[str, Any]], dict[str, str]]: + configs: dict[str, dict[str, Any]] = {} + errors: dict[str, str] = {} + if not model_ids: + return configs, errors + completed = 0 + workers = min(max(1, int(read_concurrency)), len(model_ids)) + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = { + executor.submit(discovery.get_model_config, model_id): model_id + for model_id in sorted(model_ids) + } + for future in as_completed(futures): + model_id = futures[future] + try: + config, fetch_error = future.result() + if fetch_error or not isinstance(config, dict) or not config: + errors[model_id] = str(fetch_error or "model_config_empty") + else: + configs[model_id] = dict(config) + 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] architecture_scan={completed}/{len(model_ids)} " + f"complete={len(configs)} unknown={len(errors)}" + ) + return configs, errors + + def find_certain_oom_tasks( tasks: list[OwnedTask], *, @@ -238,6 +276,94 @@ def find_certain_oom_tasks( return decisions, skipped +def find_architecture_incompatible_tasks( + tasks: list[OwnedTask], + *, + architecture_blocks: dict[str, dict[str, Any]], + task_contexts: dict[str, dict[str, Any]], + model_configs: dict[str, dict[str, Any]], +) -> tuple[list[dict[str, Any]], dict[str, int]]: + """Select waiting tasks that exactly match a learned compatibility block.""" + decisions: list[dict[str, Any]] = [] + skipped = { + "submissionContextUnknown": 0, + "submissionContextMismatch": 0, + "modelArchitectureUnknown": 0, + "noMatchingBlock": 0, + "runningMatchedProtected": 0, + } + if not architecture_blocks: + return decisions, skipped + + for task in tasks: + context = task_contexts.get(str(task.task_id)) + if not isinstance(context, dict): + skipped["submissionContextUnknown"] += 1 + continue + context_model = str(context.get("modelId") or "").strip() + context_gpu = str(context.get("targetGpu") or "").strip() + framework = str(context.get("framework") or "").strip() + task_type = str(context.get("taskType") or "").strip() + if ( + not framework + or not task_type + or (context_model and context_model != task.model_id) + or (context_gpu and context_gpu.casefold() != task.gpu_type.casefold()) + ): + skipped["submissionContextMismatch"] += 1 + continue + + profile_data = context.get("modelProfile") + if not isinstance(profile_data, dict): + profile_data = {} + model_type = profile_data.get("modelType") + architectures = profile_data.get("architectures") + if not model_type and not architectures: + config = model_configs.get(task.model_id) or {} + model_type = config.get("model_type") + architectures = config.get("architectures") + profiles = architecture_profiles(model_type, architectures) + if not profiles: + skipped["modelArchitectureUnknown"] += 1 + continue + + matching_block: dict[str, Any] | None = None + for profile in profiles: + key = architecture_compatibility_key( + task.gpu_type, + framework, + task_type, + profile["signature"], + ) + block = architecture_blocks.get(key or "") + if isinstance(block, dict): + matching_block = block + break + if matching_block is None: + skipped["noMatchingBlock"] += 1 + continue + if task.status != "waiting": + skipped["runningMatchedProtected"] += 1 + continue + decisions.append( + { + "accountIndex": task.account_index + 1, + "taskId": task.task_id, + "modelId": task.model_id, + "gpuType": task.gpu_type, + "framework": framework, + "taskType": task_type, + "status": task.status, + "architectureSignature": matching_block.get("architectureSignature"), + "architectureMatchType": matching_block.get("matchType"), + "architectureBlockExpiresAt": matching_block.get("expiresAt"), + "architectureBlockEvidenceCount": matching_block.get("evidenceCount"), + "reason": "known_framework_architecture_incompatible", + } + ) + return decisions, skipped + + def find_old_overflow_tasks( tasks: list[OwnedTask], *, @@ -319,11 +445,14 @@ def cleanup_certain_oom_tasks( read_concurrency: int = 6, stop_batch_size: int = DEFAULT_STOP_BATCH_SIZE, gpu_memory_gib: dict[str, float] | None = None, + 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 tasks and old tasks beyond each account's protected prefix.""" + """Stop deterministic OOM/architecture tasks and over-threshold old tasks.""" clients = list(modelhub.clients) reference_time = reference_time or utc_now() configured_reserved_slots = ( @@ -359,32 +488,97 @@ def cleanup_certain_oom_tasks( } model_ids = {task.model_id for task in tasks} - repository_sizes, size_errors = _load_repository_sizes( - model_ids, - discovery=discovery, - read_concurrency=read_concurrency, - log=log, - ) - oom_decisions, skipped = find_certain_oom_tasks( - tasks, - repository_sizes=repository_sizes, - gpu_memory_gib=gpu_memory_gib, - ) + repository_sizes: dict[str, int] = {} + size_errors: dict[str, str] = {} + oom_decisions: list[dict[str, Any]] = [] + skipped = { + "repositorySizeUnknown": 0, + "gpuCapacityUnknown": 0, + "fitsKnownCapacity": 0, + } + if not architecture_only: + repository_sizes, size_errors = _load_repository_sizes( + model_ids, + discovery=discovery, + read_concurrency=read_concurrency, + log=log, + ) + oom_decisions, skipped = find_certain_oom_tasks( + tasks, + repository_sizes=repository_sizes, + gpu_memory_gib=gpu_memory_gib, + ) log( f"[queue-cleanup] certain_oom={len(oom_decisions)} " f"fits={skipped['fitsKnownCapacity']} size_unknown={skipped['repositorySizeUnknown']} " f"gpu_unknown={skipped['gpuCapacityUnknown']} dry_run={str(bool(dry_run)).lower()}" ) - oom_task_keys = { - (int(decision["accountIndex"]) - 1, int(decision["taskId"])) - for decision in oom_decisions + architecture_blocks = ( + architecture_compatibility_blocks + if isinstance(architecture_compatibility_blocks, dict) + else {} + ) + task_contexts = ( + task_compatibility_contexts + if isinstance(task_compatibility_contexts, dict) + else {} + ) + block_combinations = { + ( + str(block.get("targetGpu") or "").strip().casefold(), + str(block.get("framework") or "").strip().casefold(), + str(block.get("taskType") or "").strip().casefold(), + ) + for block in architecture_blocks.values() + if isinstance(block, dict) } - # OOM tasks are stopped first. Rank the age-policy queue as it will look - # after those certain failures are gone, so an old task moving into the - # protected first N positions is not over-cancelled. + architecture_model_ids: set[str] = set() + for task in tasks: + context = task_contexts.get(str(task.task_id)) + if not isinstance(context, dict): + continue + combination = ( + task.gpu_type.casefold(), + str(context.get("framework") or "").strip().casefold(), + str(context.get("taskType") or "").strip().casefold(), + ) + profile = context.get("modelProfile") + profile = profile if isinstance(profile, dict) else {} + if combination in block_combinations and not ( + profile.get("modelType") or profile.get("architectures") + ): + architecture_model_ids.add(task.model_id) + model_configs, model_config_errors = _load_model_configs( + architecture_model_ids, + discovery=discovery, + read_concurrency=read_concurrency, + log=log, + ) + architecture_decisions, architecture_skipped = find_architecture_incompatible_tasks( + tasks, + architecture_blocks=architecture_blocks, + task_contexts=task_contexts, + model_configs=model_configs, + ) + log( + f"[queue-cleanup] architecture_incompatible={len(architecture_decisions)} " + f"blocks={len(architecture_blocks)} " + f"context_unknown={architecture_skipped['submissionContextUnknown']} " + f"architecture_unknown={architecture_skipped['modelArchitectureUnknown']} " + 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 oom_task_keys + task + for task in tasks + if (task.account_index, task.task_id) not in deterministic_task_keys ] overflow_model_ids = { task.model_id @@ -396,20 +590,32 @@ def cleanup_certain_oom_tasks( )[queue_thresholds.get(account_index, len(age_rank_tasks)):] if task.status == "waiting" } - 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), - ) + 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))} " @@ -420,10 +626,22 @@ def cleanup_certain_oom_tasks( ) decisions_by_key: dict[tuple[int, int], dict[str, Any]] = {} - for decision in oom_decisions: + for decision in [*oom_decisions, *architecture_decisions]: enriched = dict(decision) enriched["cleanupReasons"] = [decision["reason"]] - decisions_by_key[(int(decision["accountIndex"]), int(decision["taskId"]))] = enriched + key = (int(decision["accountIndex"]), int(decision["taskId"])) + existing = decisions_by_key.get(key) + if existing is None: + decisions_by_key[key] = enriched + continue + existing["cleanupReasons"].append(decision["reason"]) + existing.update( + { + field: value + for field, value in decision.items() + 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) @@ -455,15 +673,16 @@ def cleanup_certain_oom_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_oom_ids = { + planned_deterministic_ids = { int(decision["taskId"]) - for decision in oom_decisions + 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_oom_ids + 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) @@ -488,6 +707,10 @@ def cleanup_certain_oom_tasks( 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) @@ -510,6 +733,16 @@ def cleanup_certain_oom_tasks( } ) 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) @@ -520,17 +753,36 @@ def cleanup_certain_oom_tasks( if stop_failed: break decisions_by_id = {int(item["taskId"]): item for item in by_account[account_index]} - for is_oom_phase in (True, False): + for cleanup_phase in ("oom", "architecture", "age"): task_ids = sorted( task_id for task_id, decision in decisions_by_id.items() - if ("certain_oom_repository_size_exceeds_gpu_capacity" in decision["cleanupReasons"]) - == is_oom_phase + if ( + cleanup_phase == "oom" + and "certain_oom_repository_size_exceeds_gpu_capacity" + in decision["cleanupReasons"] + ) + or ( + cleanup_phase == "architecture" + and "certain_oom_repository_size_exceeds_gpu_capacity" + not in decision["cleanupReasons"] + 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 not is_oom_phase and task_ids: - # OOM stops can change actual positions, and a waiting task - # can start running after the account-wide recheck above. - # Re-read this account immediately before its age-only stop. + 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. try: phase_tasks: dict[int, OwnedTask] = {} for status in ACTIVE_FILTER_STATUSES: @@ -545,7 +797,7 @@ def cleanup_certain_oom_tasks( { "accountIndex": account_index + 1, "taskIds": task_ids, - "error": f"age_policy_final_recheck_failed: {type(exc).__name__}: {exc}", + "error": f"policy_final_recheck_failed: {type(exc).__name__}: {exc}", } ) stop_failed = True @@ -563,12 +815,24 @@ def cleanup_certain_oom_tasks( disappeared.append(decisions_by_id[task_id]) continue current_position = phase_positions.get(task_id) - if ( - account_queue_threshold is None - or current_position is None - or current_position <= account_queue_threshold - or current_task.status != "waiting" - ): + reasons = set( + decisions_by_id[task_id].get("cleanupReasons") + or [decisions_by_id[task_id].get("reason")] + ) + architecture_applies = bool( + cleanup_phase == "architecture" + 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: policy_no_longer_applies.append( { **decisions_by_id[task_id], @@ -577,7 +841,7 @@ def cleanup_certain_oom_tasks( "policyChangeReason": ( "task_started_running" if current_task.status == "running" - else "queue_position_or_status_changed" + else "queue_position_status_or_policy_changed" ), } ) @@ -612,6 +876,7 @@ def cleanup_certain_oom_tasks( return { "dryRun": bool(dry_run), + "architectureOnly": bool(architecture_only), "accounts": len(clients), "activeScanned": len(tasks), "uniqueModels": len(model_ids), @@ -622,6 +887,12 @@ def cleanup_certain_oom_tasks( "listingErrors": {str(index + 1): values for index, values in listing_errors.items()}, "certainOomCount": len(oom_decisions), "certainOomTasks": oom_decisions, + "architectureBlockCount": len(architecture_blocks), + "architectureIncompatibleCount": len(architecture_decisions), + "architectureIncompatibleTasks": architecture_decisions, + "architectureModelConfigsComplete": len(model_configs), + "architectureModelConfigErrors": model_config_errors, + "architecturePolicySkipped": architecture_skipped, "oldOverflowCount": len(old_overflow_decisions), "oldOverflowTasks": old_overflow_decisions, "oldModelQueueThresholds": [ @@ -646,7 +917,9 @@ def cleanup_certain_oom_tasks( def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Safely stop deterministic OOM and over-threshold old ModelHub tasks.") + parser = argparse.ArgumentParser( + description="Safely stop deterministic OOM/architecture and over-threshold old 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) parser.add_argument("--stop-batch-size", type=int, default=DEFAULT_STOP_BATCH_SIZE) @@ -678,6 +951,7 @@ def main(argv: list[str] | None = None) -> int: write_json(report_path, summary) print( f"[queue-cleanup] finished certain_oom={summary['certainOomCount']} " + f"architecture_incompatible={summary['architectureIncompatibleCount']} " f"old_overflow={summary['oldOverflowCount']} " f"cancelled={summary['cancelledCount']} stop_errors={len(summary['stopErrors'])} " f"report={report_path}", diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 5fff26fd..35bed5c7 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.12.2" +AGENT_VERSION = "2026.08.12.3" diff --git a/tests/test_candidate_preflight.py b/tests/test_candidate_preflight.py index a6fc1fc1..8e0954d8 100644 --- a/tests/test_candidate_preflight.py +++ b/tests/test_candidate_preflight.py @@ -492,6 +492,34 @@ class CandidatePreflightTests(unittest.TestCase): self.assertFalse(result["failureNeedsLlm"]) self.assertEqual(0, classifier.calls) + def test_fixed_platform_error_extracts_unsupported_model_type(self) -> None: + result = classify_failure_archive( + make_failure_archive( + "MODEL_NOT_SUPPORTED", + "Value error, The checkpoint you are trying to load has model type `qwen3_5` " + "but Transformers does not recognize this architecture.", + "当前框架版本不支持该模型架构，检查框架版本或改用兼容的推理后端。", + ) + ) + + self.assertEqual("framework_architecture_unsupported", result["failureCategory"]) + self.assertEqual(["qwen3_5"], result["failureUnsupportedModelTypes"]) + + def test_fixed_runtime_error_extracts_unsupported_architecture_names(self) -> None: + result = classify_failure_archive( + make_failure_archive( + "MODEL_NOT_SUPPORTED", + "ValueError: Model architectures ['CogVLMForCausalLM'] are not supported for now. " + "Supported architectures: dict_keys(['Qwen2ForCausalLM'])", + "当前框架版本不支持该模型架构，检查框架版本或改用兼容的推理后端。", + ) + ) + + self.assertEqual( + ["CogVLMForCausalLM"], + result["failureUnsupportedArchitectures"], + ) + def test_generic_unsupported_backend_does_not_create_architecture_feedback(self) -> None: classification = classify_failure_report( "ATTENTION_NOT_SUPPORTED", @@ -663,6 +691,50 @@ class CandidatePreflightTests(unittest.TestCase): self.assertEqual({}, report["architectureCompatibilityBlocks"]) + def test_parsed_model_type_builds_dynamic_block_without_saved_model_profile(self) -> None: + now = datetime.now(timezone.utc) + with tempfile.TemporaryDirectory() as temporary_dir: + tracker = OutcomeTracker(Path(temporary_dir) / "outcomes.jsonl") + tracker.record_submission( + "owner/source", + "gpu", + "vllm", + "text-generation", + "task-no-profile", + now.isoformat(), + ) + tracker._records[0].update( # noqa: SLF001 + { + "outcome": "failed", + "failureCategory": "framework_architecture_unsupported", + "failureDeterministic": True, + "failureClassificationReason": "explicit_framework_model_unsupported", + "failureUnsupportedModelTypes": ["qwen3_5"], + } + ) + report = tracker.get_stats_report() + + key = "gpu|vllm|text-generation|model_type:qwen3_5" + self.assertIn(key, report["architectureCompatibilityBlocks"]) + advisor = CandidatePreflightAdvisor(gpu_memory_gib={}) + advisor.set_feedback_stats(report) + assessment = advisor.assess( + inspection=ModelInspection( + repo_id="unrelated/repository-name", + model_config={ + "model_type": "qwen3_5", + "architectures": ["Qwen3_5ForCausalLM"], + }, + ), + task_type="text-generation", + target_gpu="gpu", + framework="vllm", + config_params="", + ) + + self.assertFalse(assessment.allowed) + self.assertEqual("preflight_learned_architecture_incompatible", assessment.reason) + def test_outcome_sync_enriches_failure_and_excludes_platform_fault_from_feedback(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: path = Path(temporary_dir) / "outcomes.jsonl" diff --git a/tests/test_poll_policy.py b/tests/test_poll_policy.py index 21bb9877..5fe128bf 100644 --- a/tests/test_poll_policy.py +++ b/tests/test_poll_policy.py @@ -1,8 +1,11 @@ from __future__ import annotations import argparse +import json import sys +import tempfile import unittest +from datetime import datetime, timezone from pathlib import Path @@ -11,10 +14,48 @@ if str(PACKAGE_DIR) in sys.path: sys.path.remove(str(PACKAGE_DIR)) sys.path.insert(0, str(PACKAGE_DIR)) -from poll_runner import resolve_age_cleanup_policy # noqa: E402 +from outcome_tracker import OutcomeTracker # noqa: E402 +from poll_runner import _load_task_compatibility_contexts, resolve_age_cleanup_policy # noqa: E402 class PollPolicyTests(unittest.TestCase): + def test_cleanup_contexts_merge_outcomes_with_older_ledger_entries(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + tracker = OutcomeTracker(root / "outcomes.jsonl") + tracker.record_submission( + "owner/new", + "gpu-a", + "vllm", + "text-generation", + "task-new", + datetime.now(timezone.utc).isoformat(), + model_profile={"architectures": ["Qwen2ForCausalLM"]}, + ) + ledger_path = root / "ledger.jsonl" + ledger_path.write_text( + json.dumps( + { + "taskId": "task-old", + "modelId": "owner/old", + "targetGpu": "gpu-b", + "framework": "mindie", + "taskType": "text-generation", + } + ) + + "\n", + encoding="utf-8", + ) + + contexts = _load_task_compatibility_contexts( + tracker, + ledger_path=ledger_path, + ) + + self.assertEqual(["Qwen2ForCausalLM"], contexts["task-new"]["modelProfile"]["architectures"]) + 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, diff --git a/tests/test_queue_cleanup.py b/tests/test_queue_cleanup.py index 73565ec8..9ea05ea4 100644 --- a/tests/test_queue_cleanup.py +++ b/tests/test_queue_cleanup.py @@ -12,10 +12,12 @@ if str(PACKAGE_DIR) in sys.path: sys.path.remove(str(PACKAGE_DIR)) sys.path.insert(0, str(PACKAGE_DIR)) +from architecture_compatibility import architecture_compatibility_key # noqa: E402 from modelhub_client import ModelHubClient, ModelHubClientPool # noqa: E402 from queue_cleanup import ( # noqa: E402 OwnedTask, cleanup_certain_oom_tasks, + find_architecture_incompatible_tasks, find_certain_oom_tasks, find_old_overflow_tasks, ) @@ -72,9 +74,11 @@ class FakeDiscovery: self, sizes: dict[str, int | None], last_modified: dict[str, datetime | None] | None = None, + configs: dict[str, dict[str, Any]] | None = None, ) -> None: self.sizes = sizes self.last_modified = last_modified or {} + self.configs = configs or {} def list_repo_tree(self, repo_id: str) -> list[dict[str, Any]]: size = self.sizes[repo_id] @@ -85,6 +89,10 @@ class FakeDiscovery: def get_model_last_modified(self, repo_id: str) -> datetime | None: return self.last_modified.get(repo_id) + def get_model_config(self, repo_id: str) -> tuple[dict[str, Any], str | None]: + config = self.configs.get(repo_id) + return (dict(config), None) if config is not None else ({}, "config_not_found") + class RecordingHttpClient: def __init__(self) -> None: @@ -103,6 +111,181 @@ class RecordingHttpClient: class QueueCleanupTests(unittest.TestCase): + @staticmethod + def architecture_block( + *, + gpu: str = "Iluvatar_bi-100", + framework: str = "vllm", + task_type: str = "text-generation", + signature: str = "architectures:qwen2forcausallm", + ) -> tuple[str, dict[str, Any]]: + key = architecture_compatibility_key(gpu, framework, task_type, signature) + assert key is not None + return key, { + "targetGpu": gpu, + "framework": framework, + "taskType": task_type, + "matchType": "architectures", + "architectureSignature": signature, + "evidenceCount": 1, + "expiresAt": "2026-09-11T00:00:00+00:00", + } + + def test_architecture_cleanup_matches_exact_context_and_protects_running(self) -> None: + key, block = self.architecture_block() + tasks = [ + OwnedTask(0, 1, "owner/waiting", "Iluvatar_bi-100", "waiting"), + OwnedTask(0, 2, "owner/running", "Iluvatar_bi-100", "running"), + OwnedTask(0, 3, "owner/other-framework", "Iluvatar_bi-100", "waiting"), + ] + contexts = { + "1": { + "modelId": "owner/waiting", + "targetGpu": "Iluvatar_bi-100", + "framework": "vllm", + "taskType": "text-generation", + "modelProfile": {"architectures": ["Qwen2ForCausalLM"]}, + }, + "2": { + "modelId": "owner/running", + "targetGpu": "Iluvatar_bi-100", + "framework": "vllm", + "taskType": "text-generation", + "modelProfile": {"architectures": ["Qwen2ForCausalLM"]}, + }, + "3": { + "modelId": "owner/other-framework", + "targetGpu": "Iluvatar_bi-100", + "framework": "mindie", + "taskType": "text-generation", + "modelProfile": {"architectures": ["Qwen2ForCausalLM"]}, + }, + } + + selected, skipped = find_architecture_incompatible_tasks( + tasks, + architecture_blocks={key: block}, + task_contexts=contexts, + model_configs={}, + ) + + self.assertEqual([1], [item["taskId"] for item in selected]) + self.assertEqual(1, skipped["runningMatchedProtected"]) + self.assertEqual(1, skipped["noMatchingBlock"]) + + def test_queue_cleanup_fetches_config_and_stops_known_incompatible_waiting_task(self) -> None: + key, block = self.architecture_block() + client = FakeQueueClient( + [ + { + "taskId": 1, + "modelId": "owner/model", + "gpuType": "Iluvatar_bi-100", + "status": "waiting", + } + ] + ) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + summary = cleanup_certain_oom_tasks( + pool, + FakeDiscovery( + {"owner/model": 1 * GIB}, + configs={ + "owner/model": { + "model_type": "qwen2", + "architectures": ["Qwen2ForCausalLM"], + } + }, + ), # type: ignore[arg-type] + architecture_compatibility_blocks={key: block}, + task_compatibility_contexts={ + "1": { + "modelId": "owner/model", + "targetGpu": "Iluvatar_bi-100", + "framework": "vllm", + "taskType": "text-generation", + "modelProfile": {}, + } + }, + log=lambda _message: None, + ) + + self.assertEqual(1, summary["architectureIncompatibleCount"]) + self.assertEqual(1, summary["cancelledCount"]) + self.assertEqual([[1]], client.stopped) + + def test_dynamic_architecture_only_cleanup_skips_expensive_size_and_age_scans(self) -> None: + key, block = self.architecture_block() + client = FakeQueueClient( + [ + { + "taskId": 1, + "modelId": "owner/model", + "gpuType": "Iluvatar_bi-100", + "status": "waiting", + } + ] + ) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + summary = cleanup_certain_oom_tasks( + pool, + FakeDiscovery({}), # type: ignore[arg-type] + architecture_compatibility_blocks={key: block}, + task_compatibility_contexts={ + "1": { + "modelId": "owner/model", + "targetGpu": "Iluvatar_bi-100", + "framework": "vllm", + "taskType": "text-generation", + "modelProfile": {"architectures": ["Qwen2ForCausalLM"]}, + } + }, + architecture_only=True, + log=lambda _message: None, + ) + + self.assertTrue(summary["architectureOnly"]) + self.assertEqual(0, summary["repositorySizesComplete"]) + self.assertEqual(0, summary["modelAgeMetadataComplete"]) + self.assertEqual(1, summary["architectureIncompatibleCount"]) + self.assertEqual([[1]], client.stopped) + + def test_architecture_cleanup_recheck_protects_task_that_started_running(self) -> None: + key, block = self.architecture_block() + client = FakeQueueClient( + [ + { + "taskId": 1, + "modelId": "owner/model", + "gpuType": "Iluvatar_bi-100", + "status": "waiting", + } + ], + promote_on_waiting_read=2, + ) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + summary = cleanup_certain_oom_tasks( + pool, + FakeDiscovery({"owner/model": 1 * GIB}), # type: ignore[arg-type] + architecture_compatibility_blocks={key: block}, + task_compatibility_contexts={ + "1": { + "modelId": "owner/model", + "targetGpu": "Iluvatar_bi-100", + "framework": "vllm", + "taskType": "text-generation", + "modelProfile": {"architectures": ["Qwen2ForCausalLM"]}, + } + }, + read_concurrency=1, + log=lambda _message: None, + ) + + self.assertEqual(1, summary["architectureIncompatibleCount"]) + self.assertEqual(0, summary["cancelledCount"]) + 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 = [