From 86088ce5774fe5c6810d77ce53c4c9f7968940f2 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Wed, 12 Aug 2026 09:00:51 +0800 Subject: [PATCH] fix: enrich cleanup from active queue context --- README.md | 8 +- modelhub_submmit_api/queue_cleanup.py | 212 ++++++++++++++++++++++---- modelhub_submmit_api/version.py | 2 +- tests/test_queue_cleanup.py | 108 +++++++++++++ 4 files changed, 297 insertions(+), 33 deletions(-) diff --git a/README.md b/README.md index bb3c7f58..f1e9ef0d 100644 --- a/README.md +++ b/README.md @@ -307,12 +307,16 @@ 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. +Version `2026.08.12.5` enriches every cleanup from the live active queue's task +level metadata. If a legacy waiting task has no recoverable framework, cleanup +queries the current ModelHub framework catalog and cancels it only when its +architecture is explicitly blocked on every listed framework. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag agent-v22 -git push origin agent-v22 +git tag agent-v23 +git push origin agent-v23 ``` diff --git a/modelhub_submmit_api/queue_cleanup.py b/modelhub_submmit_api/queue_cleanup.py index fd80e7dc..82f523fa 100644 --- a/modelhub_submmit_api/queue_cleanup.py +++ b/modelhub_submmit_api/queue_cleanup.py @@ -13,6 +13,7 @@ from common import utc_now, write_json from hf_discovery import HuggingFaceDiscovery, inspect_repo_tree from modelhub_client import ModelHubClient, ModelHubClientPool from runner_common import DEFAULT_KEY_PATH, ensure_tokens +from task_registry import task_type_from_history_task ACTIVE_FILTER_STATUSES = ("waiting", "running") @@ -26,6 +27,8 @@ class OwnedTask: model_id: str gpu_type: str status: str + task_type: str = "" + framework: str = "" def _normalize_status(value: Any) -> str: @@ -51,6 +54,10 @@ def _parse_owned_task(account_index: int, record: dict[str, Any]) -> OwnedTask | model_id=model_id, gpu_type=gpu_type, status=status, + task_type=task_type_from_history_task(record) or "", + # The current API does not expose this field today, but retaining it + # makes the cleanup automatically use it if the platform adds it. + framework=str(record.get("framework") or "").strip(), ) @@ -233,6 +240,67 @@ def _load_model_configs( return configs, errors +def _load_framework_catalog( + combinations: set[tuple[str, str]], + *, + modelhub: ModelHubClientPool, + read_concurrency: int, + log: Callable[[str], None], +) -> tuple[dict[tuple[str, str], set[str]], dict[str, str]]: + """Load the live framework set for (GPU, task type) pairs. + + Keys are normalized so active records, learned blocks, and API rows can be + compared without relying on platform capitalization. + """ + catalog: dict[tuple[str, str], set[str]] = {} + errors: dict[str, str] = {} + if not combinations: + return catalog, errors + + targets = { + (str(gpu).strip().casefold(), str(task_type).strip().casefold()): ( + str(gpu).strip(), + str(task_type).strip(), + ) + for gpu, task_type in combinations + if str(gpu).strip() and str(task_type).strip() + } + workers = min(max(1, int(read_concurrency)), max(1, len(targets))) + + def fetch(gpu: str, task_type: str) -> tuple[tuple[str, str], set[str]]: + rows = modelhub.list_framework_stats(task_type, gpu) + frameworks = { + str(row.get("framework") or "").strip() + for row in rows + if isinstance(row, dict) and str(row.get("framework") or "").strip() + } + return (gpu.casefold(), task_type.casefold()), frameworks + + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = { + executor.submit(fetch, gpu, task_type): normalized_key + for normalized_key, (gpu, task_type) in sorted(targets.items()) + } + for future in as_completed(futures): + gpu, task_type = futures[future] + key = f"{gpu}|{task_type}" + try: + returned_key, frameworks = future.result() + if frameworks: + catalog[returned_key] = frameworks + else: + errors[key] = "live_framework_catalog_empty" + except Exception as exc: + errors[key] = f"{type(exc).__name__}: {exc}" + + log( + f"[queue-cleanup] framework_catalog={len(catalog)}/{len(targets)} " + f"frameworks={sum(len(items) for items in catalog.values())} " + f"unknown={len(errors)}" + ) + return catalog, errors + + def find_certain_oom_tasks( tasks: list[OwnedTask], *, @@ -282,12 +350,16 @@ def find_architecture_incompatible_tasks( architecture_blocks: dict[str, dict[str, Any]], task_contexts: dict[str, dict[str, Any]], model_configs: dict[str, dict[str, Any]], + framework_catalog: dict[tuple[str, str], set[str]] | None = None, ) -> tuple[list[dict[str, Any]], dict[str, int]]: - """Select waiting tasks that exactly match a learned compatibility block.""" + """Select tasks with exact or safely exhaustive learned incompatibility.""" decisions: list[dict[str, Any]] = [] skipped = { "submissionContextUnknown": 0, "submissionContextMismatch": 0, + "frameworkContextUnknown": 0, + "frameworkCatalogUnknown": 0, + "partiallyBlockedFrameworkSet": 0, "modelArchitectureUnknown": 0, "noMatchingBlock": 0, "runningMatchedProtected": 0, @@ -297,21 +369,20 @@ def find_architecture_incompatible_tasks( for task in tasks: context = task_contexts.get(str(task.task_id)) - if not isinstance(context, dict): - skipped["submissionContextUnknown"] += 1 - continue + context = context if isinstance(context, dict) else {} 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() + framework = str(context.get("framework") or task.framework or "").strip() + task_type = str(context.get("taskType") or task.task_type or "").strip() if ( - not framework - or not task_type - or (context_model and context_model != task.model_id) + (context_model and context_model != task.model_id) or (context_gpu and context_gpu.casefold() != task.gpu_type.casefold()) ): skipped["submissionContextMismatch"] += 1 continue + if not task_type: + skipped["submissionContextUnknown"] += 1 + continue profile_data = context.get("modelProfile") if not isinstance(profile_data, dict): @@ -327,24 +398,57 @@ def find_architecture_incompatible_tasks( 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"], + def matching_block_for(candidate_framework: str) -> dict[str, Any] | None: + for profile in profiles: + key = architecture_compatibility_key( + task.gpu_type, + candidate_framework, + task_type, + profile["signature"], + ) + block = architecture_blocks.get(key or "") + if isinstance(block, dict): + return block + return None + + matching_blocks: dict[str, dict[str, Any]] = {} + match_scope = "exact_framework" + if framework: + matching_block = matching_block_for(framework) + if matching_block is not None: + matching_blocks[framework] = matching_block + else: + skipped["frameworkContextUnknown"] += 1 + available_frameworks = sorted( + (framework_catalog or {}).get( + (task.gpu_type.casefold(), task_type.casefold()), + set(), + ), + key=str.casefold, ) - block = architecture_blocks.get(key or "") - if isinstance(block, dict): - matching_block = block - break - if matching_block is None: + if not available_frameworks: + skipped["frameworkCatalogUnknown"] += 1 + continue + for candidate_framework in available_frameworks: + matching_block = matching_block_for(candidate_framework) + if matching_block is not None: + matching_blocks[candidate_framework] = matching_block + if len(matching_blocks) != len(available_frameworks): + if matching_blocks: + skipped["partiallyBlockedFrameworkSet"] += 1 + else: + skipped["noMatchingBlock"] += 1 + continue + framework = "*" + match_scope = "all_live_frameworks" + + if not matching_blocks: skipped["noMatchingBlock"] += 1 continue if task.status != "waiting": skipped["runningMatchedProtected"] += 1 continue + representative_block = next(iter(matching_blocks.values())) decisions.append( { "accountIndex": task.account_index + 1, @@ -354,10 +458,22 @@ def find_architecture_incompatible_tasks( "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"), + "architectureMatchScope": match_scope, + "evaluatedFrameworks": sorted(matching_blocks, key=str.casefold), + "architectureSignatures": sorted( + { + str(block.get("architectureSignature") or "") + for block in matching_blocks.values() + if block.get("architectureSignature") + } + ), + "architectureSignature": representative_block.get("architectureSignature"), + "architectureMatchType": representative_block.get("matchType"), + "architectureBlockExpiresAt": representative_block.get("expiresAt"), + "architectureBlockEvidenceCount": sum( + int(block.get("evidenceCount") or 0) + for block in matching_blocks.values() + ), "reason": "known_framework_architecture_incompatible", } ) @@ -465,7 +581,9 @@ def cleanup_certain_oom_tasks( tasks, listing_errors = collect_active_tasks(clients, read_concurrency=read_concurrency) log( f"[queue-cleanup] active_scanned={len(tasks)} accounts={len(clients)} " - f"listing_errors={sum(len(items) for items in listing_errors.values())}" + f"listing_errors={sum(len(items) for items in listing_errors.values())} " + f"task_type_recovered={sum(1 for task in tasks if task.task_type)} " + f"framework_exposed={sum(1 for task in tasks if task.framework)}" ) observed_active_counts: list[int | None] = [0 for _ in clients] @@ -533,22 +651,47 @@ def cleanup_certain_oom_tasks( for block in architecture_blocks.values() if isinstance(block, dict) } + block_gpu_task_pairs = { + (gpu, task_type) + for gpu, _framework, task_type in block_combinations + if gpu and task_type + } + framework_catalog_combinations: set[tuple[str, str]] = set() architecture_model_ids: set[str] = set() for task in tasks: context = task_contexts.get(str(task.task_id)) - if not isinstance(context, dict): + context = context if isinstance(context, dict) else {} + context_model = str(context.get("modelId") or "").strip() + context_gpu = str(context.get("targetGpu") or "").strip() + if ( + (context_model and context_model != task.model_id) + or (context_gpu and context_gpu.casefold() != task.gpu_type.casefold()) + ): continue + framework = str(context.get("framework") or task.framework or "").strip() + task_type = str(context.get("taskType") or task.task_type or "").strip() combination = ( task.gpu_type.casefold(), - str(context.get("framework") or "").strip().casefold(), - str(context.get("taskType") or "").strip().casefold(), + framework.casefold(), + task_type.casefold(), ) + gpu_task_pair = (task.gpu_type.casefold(), task_type.casefold()) profile = context.get("modelProfile") profile = profile if isinstance(profile, dict) else {} - if combination in block_combinations and not ( + exact_relevant = bool(framework and combination in block_combinations) + exhaustive_relevant = bool(not framework and gpu_task_pair in block_gpu_task_pairs) + if exhaustive_relevant: + framework_catalog_combinations.add((task.gpu_type, task_type)) + if (exact_relevant or exhaustive_relevant) and not ( profile.get("modelType") or profile.get("architectures") ): architecture_model_ids.add(task.model_id) + framework_catalog, framework_catalog_errors = _load_framework_catalog( + framework_catalog_combinations, + modelhub=modelhub, + read_concurrency=read_concurrency, + log=log, + ) model_configs, model_config_errors = _load_model_configs( architecture_model_ids, discovery=discovery, @@ -560,12 +703,16 @@ def cleanup_certain_oom_tasks( architecture_blocks=architecture_blocks, task_contexts=task_contexts, model_configs=model_configs, + framework_catalog=framework_catalog, ) 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"framework_unknown={architecture_skipped['frameworkContextUnknown']} " + f"catalog_unknown={architecture_skipped['frameworkCatalogUnknown']} " + f"partially_blocked={architecture_skipped['partiallyBlockedFrameworkSet']} " f"architecture_unknown={architecture_skipped['modelArchitectureUnknown']} " f"running_protected={architecture_skipped['runningMatchedProtected']}" ) @@ -893,6 +1040,11 @@ def cleanup_certain_oom_tasks( "architectureIncompatibleTasks": architecture_decisions, "architectureModelConfigsComplete": len(model_configs), "architectureModelConfigErrors": model_config_errors, + "architectureFrameworkCatalog": { + f"{gpu}|{task_type}": sorted(frameworks, key=str.casefold) + for (gpu, task_type), frameworks in framework_catalog.items() + }, + "architectureFrameworkCatalogErrors": framework_catalog_errors, "architecturePolicySkipped": architecture_skipped, "oldOverflowCount": len(old_overflow_decisions), "oldOverflowTasks": old_overflow_decisions, diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 828454d2..eacaee6c 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.12.4" +AGENT_VERSION = "2026.08.12.5" diff --git a/tests/test_queue_cleanup.py b/tests/test_queue_cleanup.py index 9ea05ea4..d598874a 100644 --- a/tests/test_queue_cleanup.py +++ b/tests/test_queue_cleanup.py @@ -65,6 +65,13 @@ class FakeQueueClient: self.records = [record for record in self.records if int(record["taskId"]) not in ids] return {"code": 0, "data": None} + def list_framework_stats(self, task_type: str, target_gpu: str) -> list[dict[str, Any]]: + del task_type, target_gpu + return [ + {"framework": "vllm", "modelCount": 100, "successCount": 60}, + {"framework": "transformers", "modelCount": 100, "successCount": 50}, + ] + def count_active_tasks(self, **_kwargs: Any) -> int: return len(self.records) @@ -173,6 +180,107 @@ class QueueCleanupTests(unittest.TestCase): self.assertEqual(1, skipped["runningMatchedProtected"]) self.assertEqual(1, skipped["noMatchingBlock"]) + def test_architecture_cleanup_uses_current_queue_task_type_when_every_framework_is_blocked(self) -> None: + vllm_key, vllm_block = self.architecture_block(framework="vllm") + transformers_key, transformers_block = self.architecture_block(framework="transformers") + tasks = [ + OwnedTask( + 0, + 1, + "owner/model", + "Iluvatar_bi-100", + "waiting", + task_type="text-generation", + ) + ] + + selected, skipped = find_architecture_incompatible_tasks( + tasks, + architecture_blocks={ + vllm_key: vllm_block, + transformers_key: transformers_block, + }, + task_contexts={}, + model_configs={ + "owner/model": {"architectures": ["Qwen2ForCausalLM"]}, + }, + framework_catalog={ + ("iluvatar_bi-100", "text-generation"): {"vllm", "transformers"}, + }, + ) + + self.assertEqual([1], [item["taskId"] for item in selected]) + self.assertEqual("all_live_frameworks", selected[0]["architectureMatchScope"]) + self.assertEqual(["transformers", "vllm"], selected[0]["evaluatedFrameworks"]) + self.assertEqual(1, skipped["frameworkContextUnknown"]) + + def test_architecture_cleanup_does_not_guess_when_only_some_frameworks_are_blocked(self) -> None: + key, block = self.architecture_block(framework="vllm") + tasks = [ + OwnedTask( + 0, + 1, + "owner/model", + "Iluvatar_bi-100", + "waiting", + task_type="text-generation", + ) + ] + + selected, skipped = find_architecture_incompatible_tasks( + tasks, + architecture_blocks={key: block}, + task_contexts={}, + model_configs={ + "owner/model": {"architectures": ["Qwen2ForCausalLM"]}, + }, + framework_catalog={ + ("iluvatar_bi-100", "text-generation"): {"vllm", "transformers"}, + }, + ) + + self.assertEqual([], selected) + self.assertEqual(1, skipped["partiallyBlockedFrameworkSet"]) + + def test_queue_cleanup_enriches_active_api_record_and_stops_only_all_framework_block(self) -> None: + vllm_key, vllm_block = self.architecture_block(framework="vllm") + transformers_key, transformers_block = self.architecture_block(framework="transformers") + client = FakeQueueClient( + [ + { + "taskId": 1, + "modelId": "owner/model", + "gpuType": "Iluvatar_bi-100", + "status": "waiting", + "modelTaskLevelId": 23, + "modelTaskLevel": "文本生成", + } + ] + ) + 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": {"architectures": ["Qwen2ForCausalLM"]}}, + ), # type: ignore[arg-type] + architecture_compatibility_blocks={ + vllm_key: vllm_block, + transformers_key: transformers_block, + }, + task_compatibility_contexts={}, + log=lambda _message: None, + ) + + self.assertEqual(1, summary["architectureIncompatibleCount"]) + self.assertEqual(1, summary["cancelledCount"]) + self.assertEqual( + ["transformers", "vllm"], + summary["architectureFrameworkCatalog"]["iluvatar_bi-100|text-generation"], + ) + self.assertEqual([[1]], client.stopped) + def test_queue_cleanup_fetches_config_and_stops_known_incompatible_waiting_task(self) -> None: key, block = self.architecture_block() client = FakeQueueClient(