diff --git a/README.md b/README.md index 77d3c075..bb3c7f58 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,11 @@ Optional tuning: - `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_ARCHITECTURE_COMMUNITY_PROBE_SIZE` default `50` +- `MODELHUB_ARCHITECTURE_COMMUNITY_LOOKBACK_DAYS` default `30` +- `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_DAYS` default `7` @@ -126,6 +131,18 @@ change that window. The stats report exposes `architectureCompatibilityBlocks` and per-GPU/framework block counts. The live snapshot is written to `.modelhub_state/architecture_compatibility_blacklist.json`. +Before the first cleanup/submission cycle, the poller probes the latest public +failed-validation records for usable failure archives. If the public endpoint +does not expose those archives (the current API behavior), it falls back to a +parallel, complete history scan of every configured account. Ledger metadata is +used when present; otherwise the task type is recovered from ModelHub's task +level and the selected framework is conservatively recovered from the target +container image in the failure archive. All usable historical failures are +classified once at startup and cached in `outcomes/submissions.jsonl`; subsequent +poll cycles continue incremental learning every three cycles. Empty or +incomplete public evidence is logged as a fallback, never as proof that no +architecture incompatibility exists. + Before a candidate reaches the submit queue, failure-informed preflight checks the actual ModelScope repository structure and file sizes. Non-GGUF text frameworks require root-level config, weights, and tokenizer assets. The memory @@ -286,12 +303,16 @@ 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. +Version `2026.08.12.4` bootstraps architecture feedback before the first cleanup: +it prefers recent public failure details when available, otherwise scans the +complete history of every configured account, recovers missing task/framework +metadata from task levels and target images, then continues incremental learning. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag agent-v21 -git push origin agent-v21 +git tag agent-v22 +git push origin agent-v22 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 7677ab0a..82b98190 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -103,6 +103,13 @@ bash run_poll.sh --dry-run 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. +- Startup probes recent community failures for downloadable diagnostic archives. + Because the current public task API omits them, the worker automatically scans + complete terminal history across all configured accounts instead. It joins + ledger context when available and otherwise recovers task type and conservative + framework evidence from the ModelHub task level and target container image. + This bootstrap finishes before the first queue cleanup; later failures continue + to update the same blacklist every three cycles. - 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 diff --git a/modelhub_submmit_api/failure_log_inspector.py b/modelhub_submmit_api/failure_log_inspector.py index af51151e..a6db0f19 100644 --- a/modelhub_submmit_api/failure_log_inspector.py +++ b/modelhub_submmit_api/failure_log_inspector.py @@ -32,6 +32,28 @@ MODEL_ARCHITECTURES_NOT_SUPPORTED_PATTERN = re.compile( r"(?:are|is)\s+not\s+supported\s+for\s+now", re.IGNORECASE, ) +SUBMITTED_FRAMEWORK_PATTERN = re.compile( + r"\[submit\]\s*framework\s*:\s*(?P[A-Za-z0-9_.-]{1,80})", + re.IGNORECASE, +) +TARGET_DOCKER_IMAGE_PATTERN = re.compile( + r"\[submit\]\s*docker_image\s*:\s*(?P\S{1,500})", + re.IGNORECASE, +) +FRAMEWORK_IMAGE_MARKERS = ( + ("vllm-customized", ("vllm-customized", "vllm_customized")), + ("vllm_fix_tokenizer", ("vllm-fix-tokenizer", "vllm_fix_tokenizer")), + ("sentence-transformers", ("sentence-transformers", "sentence_transformers")), + ("sherpa-onnx", ("sherpa-onnx", "sherpa_onnx")), + ("llamacpp", ("llamacpp", "llama-cpp", "llama.cpp")), + ("vllm-mlu", ("vllm-mlu", "vllm_mlu")), + ("vllm-016", ("vllm-016", "vllm_016")), + ("diffusers", ("diffusers",)), + ("transformers", ("transformers",)), + ("sglang", ("sglang",)), + ("funasr", ("funasr",)), + ("vllm", ("vllm",)), +) def fetch_and_classify_failure_log( @@ -111,6 +133,10 @@ def classify_failure_archive( result["failureUnsupportedArchitectures"] = unsupported_architectures if unsupported_model_types: result["failureUnsupportedModelTypes"] = unsupported_model_types + detected_framework, framework_source = _extract_submitted_framework(runtime_log) + if detected_framework: + result["failureDetectedFramework"] = detected_framework + result["failureDetectedFrameworkSource"] = framework_source 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 {}), @@ -137,6 +163,22 @@ def classify_failure_archive( return result +def _extract_submitted_framework(runtime_log: str) -> tuple[str | None, str | None]: + """Recover only explicit or unambiguous target-framework evidence.""" + explicit = SUBMITTED_FRAMEWORK_PATTERN.search(runtime_log) + if explicit: + return explicit.group("framework").strip(), "submit_framework" + + image_match = TARGET_DOCKER_IMAGE_PATTERN.search(runtime_log) + if image_match is None: + return None, None + target_image = image_match.group("image").strip().casefold() + for framework, markers in FRAMEWORK_IMAGE_MARKERS: + if any(marker in target_image for marker in markers): + return framework, "target_docker_image" + return None, None + + def _extract_error_lines(runtime_log: str) -> list[str]: selected: list[str] = [] for raw_line in runtime_log.splitlines(): diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index f033a93f..6d329a0d 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -155,6 +155,9 @@ class ModelHubClient: end_time: datetime | None = None, gpu_type: str | None = None, model_id: str | None = None, + status: str | None = None, + verify_result: int | None = None, + max_records: int = 0, ) -> list[dict[str, Any]]: current = 1 records: list[dict[str, Any]] = [] @@ -167,10 +170,14 @@ class ModelHubClient: end_time=end_time, gpu_type=gpu_type, model_id=model_id, + status=status, + verify_result=verify_result, ) page_data = page.get("data") or {} page_records = page_data.get("records") or [] records.extend(page_records) + if max_records > 0 and len(records) >= max_records: + return records[:max_records] pages = int(page_data.get("pages") or 0) if pages <= current or not page_records: break diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index 156232a1..0fcca09f 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -20,12 +20,14 @@ from failure_log_inspector import fetch_and_classify_failure_log from history_stats import classify_failure, is_failure, is_success from llm_classifier import LLMAssistedClassifier from modelhub_client import ModelHubClient, ModelHubClientPool +from task_registry import task_type_from_history_task DEFAULT_OUTCOMES_PATH = Path("outcomes/submissions.jsonl") FAILURE_ENRICHMENT_LIMIT = 40 FAILURE_ENRICHMENT_WORKERS = 4 FAILURE_ENRICHMENT_MAX_ATTEMPTS = 3 +TERMINAL_TASK_STATUSES = {"success", "failed", "error", "cancelled", "completed"} def _now_iso() -> str: @@ -188,7 +190,7 @@ class OutcomeTracker: updated_count += 1 else: status = str(task.get("status") or "").lower() - if status in {"success", "failed", "error", "cancelled", "completed"}: + if status in TERMINAL_TASK_STATUSES: record = self._create_record_from_task(task) self._records.append(record) self._by_task_id[task_id] = record @@ -197,10 +199,9 @@ 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. 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. + # Retry a small bounded set of our own failed submissions. Newer logs + # contain the target image, so enrichment can recover the framework even + # when the history API omits it. candidate_ids = {id(record) for record in enrichment_candidates} for record in self._records: if len(enrichment_candidates) >= FAILURE_ENRICHMENT_LIMIT: @@ -210,7 +211,6 @@ class OutcomeTracker: if ( record.get("outcome") == "failed" and record.get("logCosUrl") - and record.get("framework") and not record.get("failureCategory") and int(record.get("failureEnrichmentAttempts") or 0) < FAILURE_ENRICHMENT_MAX_ATTEMPTS ): @@ -228,7 +228,185 @@ class OutcomeTracker: return updated_count - def _enrich_failure_records(self, records: list[dict[str, Any]]) -> int: + def bootstrap_from_history_tasks( + self, + tasks: list[dict[str, Any]], + *, + task_contexts: dict[str, dict[str, Any]] | None = None, + enrichment_limit: int = 0, + enrichment_workers: int = 8, + log: Any = None, + ) -> dict[str, int]: + """Import terminal history and classify every usable historical failure.""" + contexts = task_contexts if isinstance(task_contexts, dict) else {} + imported = 0 + refreshed = 0 + terminal_seen = 0 + seen_task_ids: set[str] = set() + + for raw_task in tasks: + if not isinstance(raw_task, dict): + continue + task_id_value = raw_task.get("taskId") + if task_id_value is None: + continue + task_id = str(task_id_value) + if not task_id or task_id in seen_task_ids: + continue + seen_task_ids.add(task_id) + status = str(raw_task.get("status") or "").strip().lower() + if status not in TERMINAL_TASK_STATUSES: + continue + terminal_seen += 1 + + task = self._enrich_history_task(raw_task, contexts.get(task_id)) + existing = self._by_task_id.get(task_id) + if existing is None: + record = self._create_record_from_task(task) + self._records.append(record) + self._by_task_id[task_id] = record + self._by_model_gpu[(record["modelId"], record["targetGpu"])].append(record) + imported += 1 + continue + + self._merge_record_metadata(existing, task) + if existing.get("outcome") in {"pending", "policy_cancelled"}: + self._update_record_from_task(existing, task) + refreshed += 1 + elif task.get("logCosUrl") and not existing.get("failureCategory"): + existing["logCosUrl"] = task.get("logCosUrl") + + candidates = [ + record + for record in self._records + if ( + record.get("outcome") == "failed" + and record.get("logCosUrl") + and not record.get("failureCategory") + and int(record.get("failureEnrichmentAttempts") or 0) + < FAILURE_ENRICHMENT_MAX_ATTEMPTS + ) + ] + candidates.sort(key=_outcome_record_timestamp, reverse=True) + if enrichment_limit > 0: + candidates = candidates[:enrichment_limit] + self._last_sync_time = utc_now() + self._rebuild_failed_index() + if imported or refreshed: + self.save() + + candidate_task_ids = [ + str(record["taskId"]) + for record in candidates + if record.get("taskId") is not None + ] + enrichment_attempts = 0 + classified_total = 0 + errors_total = 0 + batch_size = 200 + total_candidates = len(candidate_task_ids) + for offset in range(0, total_candidates, batch_size): + batch_ids = candidate_task_ids[offset : offset + batch_size] + batch_records = [ + self._by_task_id[task_id] + for task_id in batch_ids + if task_id in self._by_task_id + ] + + def progress( + completed: int, + _total: int, + classified: int, + errors: int, + ) -> None: + if log is None: + return + global_completed = offset + completed + if global_completed == total_candidates or global_completed % 50 == 0: + log( + f"[architecture-bootstrap] failure_logs={global_completed}/{total_candidates} " + f"classified={classified_total + classified} " + f"errors={errors_total + errors}" + ) + + enrichment_attempts += self._enrich_failure_records( + batch_records, + workers=max(1, int(enrichment_workers)), + progress=progress, + ) + classified_total += sum( + 1 for record in batch_records if record.get("failureCategory") + ) + errors_total += sum( + 1 for record in batch_records if record.get("failureEnrichmentError") + ) + self._rebuild_failed_index() + self.save() + + explicit_architecture_failures = sum( + 1 for record in self._records if _is_explicit_architecture_failure(record) + ) + recovered_frameworks = sum( + 1 + for record in self._records + if record.get("failureDetectedFramework") and record.get("framework") + ) + return { + "recordsScanned": len(seen_task_ids), + "terminalRecords": terminal_seen, + "importedRecords": imported, + "refreshedRecords": refreshed, + "eligibleFailureLogs": total_candidates, + "enrichmentAttempts": enrichment_attempts, + "explicitArchitectureFailures": explicit_architecture_failures, + "recoveredFrameworks": recovered_frameworks, + } + + @staticmethod + def _enrich_history_task( + task: dict[str, Any], + context: dict[str, Any] | None, + ) -> dict[str, Any]: + enriched = dict(task) + context = context if isinstance(context, dict) else {} + if not enriched.get("modelId") and context.get("modelId"): + enriched["modelId"] = context.get("modelId") + if not (enriched.get("gpuType") or enriched.get("targetGpu")) and context.get("targetGpu"): + enriched["targetGpu"] = context.get("targetGpu") + if not enriched.get("framework") and context.get("framework"): + enriched["framework"] = context.get("framework") + if not enriched.get("taskType"): + enriched["taskType"] = context.get("taskType") or task_type_from_history_task(enriched) + if context.get("modelProfile") and not enriched.get("modelProfile"): + enriched["modelProfile"] = context.get("modelProfile") + if context.get("submitTime") and not ( + enriched.get("submitTime") or enriched.get("createTime") + ): + enriched["submitTime"] = context.get("submitTime") + return enriched + + @staticmethod + def _merge_record_metadata(record: dict[str, Any], task: dict[str, Any]) -> None: + field_sources = { + "modelId": task.get("modelId") or task.get("model_id"), + "targetGpu": task.get("gpuType") or task.get("targetGpu"), + "framework": task.get("framework"), + "taskType": task.get("taskType") or task_type_from_history_task(task), + "modelProfile": task.get("modelProfile"), + } + for field, value in field_sources.items(): + if not record.get(field) and value not in (None, "", "unknown"): + record[field] = dict(value) if field == "modelProfile" and isinstance(value, dict) else value + if task.get("logCosUrl") and not record.get("failureCategory"): + record["logCosUrl"] = task.get("logCosUrl") + + def _enrich_failure_records( + self, + records: list[dict[str, Any]], + *, + workers: int = FAILURE_ENRICHMENT_WORKERS, + progress: Any = None, + ) -> int: if not records: return 0 @@ -244,19 +422,29 @@ class OutcomeTracker: return record, None, f"{type(exc).__name__}: {exc}" attempted = 0 - with ThreadPoolExecutor(max_workers=min(FAILURE_ENRICHMENT_WORKERS, len(records))) as executor: + classified = 0 + errors = 0 + with ThreadPoolExecutor(max_workers=min(max(1, int(workers)), len(records))) as executor: futures = [executor.submit(inspect, record) for record in records] for future in as_completed(futures): record, result, error = future.result() attempted += 1 record["failureEnrichmentAttempts"] = int(record.get("failureEnrichmentAttempts") or 0) + 1 if result is None: + errors += 1 record["failureEnrichmentError"] = error + if progress is not None: + progress(attempted, len(records), classified, errors) continue record.update(result) + if not record.get("framework") and result.get("failureDetectedFramework"): + record["framework"] = result.get("failureDetectedFramework") record["failReason"] = result.get("failureCategory") or record.get("failReason") record["failureEnrichmentError"] = None record.pop("logCosUrl", None) + classified += 1 + if progress is not None: + progress(attempted, len(records), classified, errors) return attempted def is_model_gpu_failed( @@ -438,7 +626,7 @@ class OutcomeTracker: continue model_id = record.get("modelId") or "" target_gpu = record.get("targetGpu") or "" - event_time = parse_datetime(record.get("lastSyncTime")) or parse_datetime(record.get("submitTime")) + event_time = parse_datetime(record.get("submitTime")) or parse_datetime(record.get("lastSyncTime")) if not model_id or not target_gpu or event_time is None: continue key = (model_id, target_gpu) @@ -478,12 +666,17 @@ class OutcomeTracker: @staticmethod def _create_record_from_task(task: dict[str, Any]) -> dict[str, Any]: - create_time = parse_datetime(task.get("createTime")) + create_time = ( + parse_datetime(task.get("submitTime")) + or parse_datetime(task.get("createTime")) + or parse_datetime(task.get("updateTime")) + ) + model_profile = task.get("modelProfile") record: dict[str, Any] = { "modelId": task.get("modelId") or task.get("model_id") or "", "targetGpu": task.get("gpuType") or task.get("targetGpu") or "", "framework": task.get("framework") or "", - "taskType": task.get("taskType") or "", + "taskType": task.get("taskType") or task_type_from_history_task(task) or "", "taskId": str(task.get("taskId")) if task.get("taskId") is not None else None, "submitTime": create_time.isoformat() if create_time else _now_iso(), "lastSyncTime": _now_iso(), @@ -492,6 +685,7 @@ class OutcomeTracker: "outcome": "pending", "failReason": None, "logCosUrl": task.get("logCosUrl"), + "modelProfile": dict(model_profile) if isinstance(model_profile, dict) else {}, } if is_success(task): record["outcome"] = "success" @@ -725,13 +919,13 @@ def _latest_platform_failure_at(records: list[dict[str, Any]]) -> str | None: for record in records: if not _is_platform_failure(record): continue - timestamp = parse_datetime(record.get("lastSyncTime")) or parse_datetime(record.get("submitTime")) + timestamp = parse_datetime(record.get("submitTime")) or parse_datetime(record.get("lastSyncTime")) return timestamp.isoformat() if timestamp else None return None def _outcome_record_timestamp(record: dict[str, Any]) -> float: - timestamp = parse_datetime(record.get("lastSyncTime")) or parse_datetime(record.get("submitTime")) + timestamp = parse_datetime(record.get("submitTime")) or parse_datetime(record.get("lastSyncTime")) return timestamp.timestamp() if timestamp else 0.0 diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 491f8447..d48ffa6e 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -5,6 +5,8 @@ import json import os import sys import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import timedelta from pathlib import Path from typing import Any, Callable @@ -36,6 +38,14 @@ DEFAULT_ARCHITECTURE_BLACKLIST_PATH = Path( ) +def _env_int(name: str, default: int, *, minimum: int = 0, maximum: int = 100_000) -> int: + try: + value = int(os.getenv(name, str(default))) + except ValueError: + value = default + return min(maximum, max(minimum, value)) + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Continuously poll ModelHub queue slots and refill submissions.") parser.add_argument("--daily-target", type=int, default=0, help="Total submissions to aim for per UTC day; 0 means unlimited") @@ -309,6 +319,175 @@ def _load_task_compatibility_contexts( return contexts +def _load_complete_owned_history( + modelhub_client: ModelHubClient | ModelHubClientPool, +) -> tuple[list[dict[str, Any]], list[int]]: + clients = ( + list(modelhub_client.clients) + if isinstance(modelhub_client, ModelHubClientPool) + else [modelhub_client] + ) + by_account: dict[int, list[dict[str, Any]]] = {} + errors: list[int] = [] + with ThreadPoolExecutor(max_workers=min(12, max(1, len(clients)))) as executor: + futures = { + executor.submit(client.list_tasks, page_size=100, only_mine=True): index + for index, client in enumerate(clients, start=1) + } + for future in as_completed(futures): + account_index = futures[future] + try: + by_account[account_index] = future.result() + except Exception: + errors.append(account_index) + + deduped: dict[str, dict[str, Any]] = {} + anonymous: list[dict[str, Any]] = [] + for account_index in sorted(by_account): + for task in by_account[account_index]: + if not isinstance(task, dict): + continue + task_id = task.get("taskId") + if task_id is None: + anonymous.append(task) + continue + deduped[str(task_id)] = task + return [*deduped.values(), *anonymous], sorted(errors) + + +def _bootstrap_architecture_history( + *, + modelhub_client: ModelHubClient | ModelHubClientPool, + outcome_tracker: OutcomeTracker, + ledger_path: Path, + now, +) -> dict[str, Any]: + """Prefer public failure evidence, then fall back to all owned history.""" + probe_size = _env_int( + "MODELHUB_ARCHITECTURE_COMMUNITY_PROBE_SIZE", + 50, + minimum=1, + maximum=100, + ) + community_records: list[dict[str, Any]] = [] + community_probe_error: str | None = None + try: + probe_payload = modelhub_client.list_tasks_page( + current=1, + page_size=probe_size, + only_mine=False, + status="success", + verify_result=-1, + ) + probe_records = (probe_payload.get("data") or {}).get("records") or [] + community_records = [record for record in probe_records if isinstance(record, dict)] + except Exception as exc: + community_probe_error = f"{type(exc).__name__}: {exc}" + + community_usable = [ + record + for record in community_records + if record.get("logCosUrl") + and record.get("modelId") + and record.get("gpuType") + ] + log( + f"[architecture-bootstrap] community_probe={len(community_records)} " + f"usable_failure_details={len(community_usable)} " + f"error={'none' if community_probe_error is None else community_probe_error}" + ) + + source = "community_latest" + listing_errors: list[int] = [] + if community_usable: + lookback_days = _env_int( + "MODELHUB_ARCHITECTURE_COMMUNITY_LOOKBACK_DAYS", + 30, + minimum=1, + maximum=365, + ) + latest_limit = _env_int( + "MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT", + 5000, + minimum=1, + maximum=50_000, + ) + try: + history_tasks = modelhub_client.list_tasks( + page_size=100, + only_mine=False, + begin_time=now - timedelta(days=lookback_days), + end_time=now, + status="success", + verify_result=-1, + max_records=latest_limit, + ) + history_tasks = [task for task in history_tasks if task.get("logCosUrl")] + log( + f"[architecture-bootstrap] source=community_latest " + f"lookback_days={lookback_days} records={len(history_tasks)} limit={latest_limit}" + ) + except Exception as exc: + source = "owned_full_history" + log( + f"[architecture-bootstrap] community_history_error={type(exc).__name__}: {exc} " + "fallback=owned_full_history" + ) + history_tasks, listing_errors = _load_complete_owned_history(modelhub_client) + else: + source = "owned_full_history" + history_tasks, listing_errors = _load_complete_owned_history(modelhub_client) + account_count = ( + len(modelhub_client.clients) + if isinstance(modelhub_client, ModelHubClientPool) + else 1 + ) + log( + f"[architecture-bootstrap] source=owned_full_history records={len(history_tasks)} " + f"accounts={account_count} listing_errors={','.join(map(str, listing_errors)) or 'none'}" + ) + + contexts = _load_task_compatibility_contexts( + outcome_tracker, + ledger_path=ledger_path, + ) + summary = outcome_tracker.bootstrap_from_history_tasks( + history_tasks, + task_contexts=contexts, + enrichment_limit=_env_int( + "MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS", + 0, + minimum=0, + maximum=100_000, + ), + enrichment_workers=_env_int( + "MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS", + 8, + minimum=1, + maximum=16, + ), + log=log, + ) + feedback = outcome_tracker.get_stats_report() + block_count = len(feedback.get("architectureCompatibilityBlocks") or {}) + summary.update( + { + "source": source, + "communityProbeRecords": len(community_records), + "communityUsableFailureDetails": len(community_usable), + "listingErrorAccounts": listing_errors, + "architectureBlocks": block_count, + } + ) + log( + f"[architecture-bootstrap] finished source={source} " + f"terminal={summary['terminalRecords']} failure_logs={summary['eligibleFailureLogs']} " + f"explicit_architecture_failures={summary['explicitArchitectureFailures']} " + f"recovered_frameworks={summary['recoveredFrameworks']} blocks={block_count}" + ) + return summary + + def run_poll_loop( *, base_args: argparse.Namespace, @@ -338,6 +517,38 @@ def run_poll_loop( f"poll_interval={base_args.poll_interval_seconds}s idle_interval={base_args.idle_interval_seconds}s" ) + architecture_bootstrap_summary: dict[str, Any] = {"enabled": False} + if not getattr(base_args, "skip_outcome_sync", False): + try: + architecture_bootstrap_summary = _bootstrap_architecture_history( + modelhub_client=modelhub_client, + outcome_tracker=outcome_tracker, + ledger_path=Path(base_args.ledger_path), + now=now, + ) + architecture_bootstrap_summary["enabled"] = True + _persist_architecture_blacklist( + outcome_tracker.get_stats_report(), + path=Path( + getattr( + base_args, + "architecture_blacklist_path", + DEFAULT_ARCHITECTURE_BLACKLIST_PATH, + ) + ), + ) + except Exception as exc: + architecture_bootstrap_summary = { + "enabled": True, + "error": f"{type(exc).__name__}: {exc}", + } + log( + f"[architecture-bootstrap] error={type(exc).__name__}: {exc} " + "continue_polling=true" + ) + else: + architecture_bootstrap_summary["reason"] = "outcome_sync_disabled" + cycle_summaries: list[dict[str, Any]] = [] queue_cleanup_runs: list[dict[str, Any]] = [] submitted_total = 0 @@ -608,6 +819,7 @@ def run_poll_loop( "pollRunDir": str(poll_run_dir), "cycleSummaries": cycle_summaries, "queueCleanupRuns": queue_cleanup_runs, + "architectureBootstrap": architecture_bootstrap_summary, "outcomeStats": stats_report, } write_json(poll_run_dir / "summary.json", summary) diff --git a/modelhub_submmit_api/queue_cleanup.py b/modelhub_submmit_api/queue_cleanup.py index 89f34459..fd80e7dc 100644 --- a/modelhub_submmit_api/queue_cleanup.py +++ b/modelhub_submmit_api/queue_cleanup.py @@ -564,6 +564,7 @@ def cleanup_certain_oom_tasks( log( f"[queue-cleanup] architecture_incompatible={len(architecture_decisions)} " f"blocks={len(architecture_blocks)} " + f"rule_state={'ready' if architecture_blocks else 'no_learned_blocks'} " f"context_unknown={architecture_skipped['submissionContextUnknown']} " f"architecture_unknown={architecture_skipped['modelArchitectureUnknown']} " f"running_protected={architecture_skipped['runningMatchedProtected']}" diff --git a/modelhub_submmit_api/success_history_runner.py b/modelhub_submmit_api/success_history_runner.py index 6dcd15dc..c0f001d4 100644 --- a/modelhub_submmit_api/success_history_runner.py +++ b/modelhub_submmit_api/success_history_runner.py @@ -13,40 +13,10 @@ from modelhub_client import ModelHubClient, ModelHubClientPool from models import HFModelSummary from poll_runner import DEFAULT_POLL_RUNS_DIR, run_poll_loop from runner_common import DEFAULT_KEY_PATH, ensure_tokens -from task_registry import TASK_SPEC_BY_TYPE +from task_registry import TASK_SPEC_BY_TYPE, task_type_from_history_task from template_selector import TemplateSelector -TASK_TYPE_BY_MODEL_TASK_LEVEL_ID = { - "0": "text-generation", - "2": "visual-multi-modal", - "21": "text-to-image-generation", - "23": "text-generation", - "29": "feature_emb", - "34": "feature_emb", - "39": "question_answering", - "54": "asr", - "116": "vision_classification", - "198": "vision_classification", - "207": "reinforcement_learning", - "213": "text-to-image-generation", -} - -TASK_TYPE_BY_MODEL_TASK_LEVEL = { - "文本生成": "text-generation", - "视觉多模态理解": "visual-multi-modal", - "图片生成图片": "text-to-image-generation", - "文生图": "text-to-image-generation", - "特征抽取": "feature_emb", - "句子相似度": "feature_emb", - "问答": "question_answering", - "语音识别": "asr", - "视觉分类": "vision_classification", - "零样本图像分类": "vision_classification", - "强化学习": "reinforcement_learning", -} - - def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Continuously refill submissions, but only on GPUs with historical verified successes." @@ -137,21 +107,6 @@ def pipeline_tag_for_task_type(task_type: str | None) -> str | None: return spec.pipeline_tags[0] -def task_type_from_history_task(task: dict[str, Any]) -> str | None: - task_type = task.get("taskType") - if task_type and str(task_type) in TASK_SPEC_BY_TYPE: - return str(task_type) - level_id = task.get("modelTaskLevelId") - if level_id is not None: - mapped = TASK_TYPE_BY_MODEL_TASK_LEVEL_ID.get(str(level_id)) - if mapped: - return mapped - level_name = task.get("modelTaskLevel") - if level_name: - return TASK_TYPE_BY_MODEL_TASK_LEVEL.get(str(level_name)) - return None - - def successful_models_from_tasks(tasks: list[dict[str, Any]], *, limit: int) -> tuple[list[HFModelSummary], dict[str, set[str]]]: by_model: dict[str, HFModelSummary] = {} gpus_by_model: dict[str, set[str]] = {} diff --git a/modelhub_submmit_api/task_registry.py b/modelhub_submmit_api/task_registry.py index 88175ed2..0d286c5f 100644 --- a/modelhub_submmit_api/task_registry.py +++ b/modelhub_submmit_api/task_registry.py @@ -44,6 +44,52 @@ TASK_SPECS: tuple[TaskSpec, ...] = ( TASK_SPEC_BY_TYPE = {task.task_type: task for task in TASK_SPECS} +TASK_TYPE_BY_MODEL_TASK_LEVEL_ID = { + "0": "text-generation", + "2": "visual-multi-modal", + "21": "text-to-image-generation", + "23": "text-generation", + "29": "feature_emb", + "34": "feature_emb", + "39": "question_answering", + "54": "asr", + "116": "vision_classification", + "198": "vision_classification", + "207": "reinforcement_learning", + "213": "text-to-image-generation", +} + +TASK_TYPE_BY_MODEL_TASK_LEVEL = { + "文本生成": "text-generation", + "视觉多模态理解": "visual-multi-modal", + "图片生成图片": "text-to-image-generation", + "文生图": "text-to-image-generation", + "特征抽取": "feature_emb", + "句子相似度": "feature_emb", + "问答": "question_answering", + "语音识别": "asr", + "视觉分类": "vision_classification", + "零样本图像分类": "vision_classification", + "强化学习": "reinforcement_learning", +} + + +def task_type_from_history_task(task: dict) -> str | None: + """Recover the API task type from current or legacy history fields.""" + task_type = task.get("taskType") + if task_type and str(task_type) in TASK_SPEC_BY_TYPE: + return str(task_type) + level_id = task.get("modelTaskLevelId") + if level_id is not None: + mapped = TASK_TYPE_BY_MODEL_TASK_LEVEL_ID.get(str(level_id)) + if mapped: + return mapped + level_name = task.get("modelTaskLevel") + if level_name: + return TASK_TYPE_BY_MODEL_TASK_LEVEL.get(str(level_name)) + return None + + def all_task_types() -> list[str]: return [task.task_type for task in TASK_SPECS] diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 35bed5c7..828454d2 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.12.3" +AGENT_VERSION = "2026.08.12.4" diff --git a/tests/test_candidate_preflight.py b/tests/test_candidate_preflight.py index 8e0954d8..8336245b 100644 --- a/tests/test_candidate_preflight.py +++ b/tests/test_candidate_preflight.py @@ -574,6 +574,60 @@ class CandidatePreflightTests(unittest.TestCase): self.assertFalse(result["failureNeedsLlm"]) self.assertEqual(1, classifier.calls) + def test_failure_archive_recovers_framework_from_target_docker_image(self) -> None: + result = classify_failure_archive( + make_failure_archive( + "MODEL_NOT_SUPPORTED", + "\n".join( + [ + "[submit]docker_image: registry/enginex-sunrise/enginex-s2-vllm:v1", + "model type `qwen3_5` but Transformers does not recognize this architecture", + ] + ), + "请换用支持的模型", + ) + ) + + self.assertEqual("vllm", result["failureDetectedFramework"]) + self.assertEqual("target_docker_image", result["failureDetectedFrameworkSource"]) + self.assertEqual(["qwen3_5"], result["failureUnsupportedModelTypes"]) + + def test_history_bootstrap_builds_block_without_api_framework_field(self) -> None: + now = datetime.now(timezone.utc) + task = { + "taskId": "history-failure", + "modelId": "owner/history-model", + "gpuType": "Biren_166m", + "modelTaskLevelId": 23, + "status": "success", + "verifyResult": -1, + "updateTime": now.isoformat(), + "logCosUrl": "https://logs.invalid/history-failure.zip", + } + classification = { + "failureCategory": "framework_architecture_unsupported", + "failureScope": "model_gpu_framework", + "failureAction": "block_gpu_framework_architecture", + "failureDeterministic": True, + "failureClassificationReason": "explicit_framework_model_unsupported", + "failureUnsupportedModelTypes": ["qwen3_5"], + "failureDetectedFramework": "vllm", + "failureDetectedFrameworkSource": "target_docker_image", + } + with tempfile.TemporaryDirectory() as temporary_dir: + tracker = OutcomeTracker(Path(temporary_dir) / "outcomes.jsonl") + with patch( + "outcome_tracker.fetch_and_classify_failure_log", + return_value=classification, + ): + summary = tracker.bootstrap_from_history_tasks([task]) + report = tracker.get_stats_report() + + key = "biren_166m|vllm|text-generation|model_type:qwen3_5" + self.assertEqual(1, summary["enrichmentAttempts"]) + self.assertEqual(1, summary["recoveredFrameworks"]) + self.assertIn(key, report["architectureCompatibilityBlocks"]) + def test_explicit_failure_learns_exact_gpu_framework_architecture_block(self) -> None: now = datetime.now(timezone.utc) with tempfile.TemporaryDirectory() as temporary_dir: diff --git a/tests/test_poll_policy.py b/tests/test_poll_policy.py index 5fe128bf..9d779f10 100644 --- a/tests/test_poll_policy.py +++ b/tests/test_poll_policy.py @@ -15,10 +15,61 @@ if str(PACKAGE_DIR) in sys.path: sys.path.insert(0, str(PACKAGE_DIR)) from outcome_tracker import OutcomeTracker # noqa: E402 -from poll_runner import _load_task_compatibility_contexts, resolve_age_cleanup_policy # noqa: E402 +from poll_runner import ( # noqa: E402 + _bootstrap_architecture_history, + _load_task_compatibility_contexts, + resolve_age_cleanup_policy, +) class PollPolicyTests(unittest.TestCase): + def test_architecture_bootstrap_falls_back_when_public_logs_are_hidden(self) -> None: + class HistoryClient: + @staticmethod + def list_tasks_page(**_kwargs): # noqa: ANN003 + return { + "data": { + "records": [ + { + "taskId": "public-failure", + "modelId": "public/model", + "gpuType": "gpu", + "status": "success", + "verifyResult": -1, + "logCosUrl": None, + } + ] + } + } + + @staticmethod + def list_tasks(**_kwargs): # noqa: ANN003 + return [ + { + "taskId": "owned-success", + "modelId": "owner/model", + "gpuType": "gpu", + "modelTaskLevelId": 23, + "status": "success", + "verifyResult": 1, + "updateTime": datetime.now(timezone.utc).isoformat(), + } + ] + + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + tracker = OutcomeTracker(root / "outcomes.jsonl") + summary = _bootstrap_architecture_history( + modelhub_client=HistoryClient(), # type: ignore[arg-type] + outcome_tracker=tracker, + ledger_path=root / "ledger.jsonl", + now=datetime.now(timezone.utc), + ) + + self.assertEqual("owned_full_history", summary["source"]) + self.assertEqual(1, summary["terminalRecords"]) + self.assertEqual(0, summary["communityUsableFailureDetails"]) + def test_cleanup_contexts_merge_outcomes_with_older_ledger_entries(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: root = Path(temporary_dir)