From ccae7ff8f31b271c7664c9e4043d129afe69c336 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Sun, 2 Aug 2026 18:48:47 +0800 Subject: [PATCH] fix: prevent repeated model GPU submissions --- README.md | 10 +- modelhub_submmit_api/README.md | 5 + modelhub_submmit_api/daily_runner.py | 14 +- modelhub_submmit_api/main.py | 116 ++++++++- modelhub_submmit_api/modelhub_client.py | 69 ++++-- modelhub_submmit_api/poll_runner.py | 5 + modelhub_submmit_api/submission_exclusions.py | 67 ++++++ modelhub_submmit_api/version.py | 2 +- tests/test_submission_safety.py | 225 ++++++++++++++++++ 9 files changed, 489 insertions(+), 24 deletions(-) create mode 100644 modelhub_submmit_api/submission_exclusions.py create mode 100644 tests/test_submission_safety.py diff --git a/README.md b/README.md index 9a0d5e80..ec7f7873 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ Optional tuning: - `MODELHUB_AGENT_RESERVATION_TTL_SECONDS` default `120` - `MODELHUB_AGENT_INSTANCE_ID` optional stable worker identity used to spread concurrent agents across accounts and candidates - `MODELHUB_AGENT_CLAIMS_PATH` default `.modelhub_state/submission_claims.jsonl` +- `MODELHUB_SUBMISSION_EXCLUSIONS_PATH` default `.modelhub_state/submission_exclusions.jsonl` - `MODELHUB_AGENT_DAILY_TARGET` - `MODELHUB_AGENT_MIN_DOWNLOADS` - `MODELHUB_AGENT_GPUS` @@ -77,6 +78,12 @@ locally failed model/GPU pair cools down for 24 hours instead of being excluded forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done` includes `skip_reasons` so an empty candidate pool is directly diagnosable. +Community deduplication is scoped to the exact model/GPU combination. A model +adapted on one GPU remains eligible for another GPU. Before each submission the +worker performs an uncached community check; a lookup failure defers the task +instead of failing open. Platform uniqueness rejections are persisted per +model/GPU in `.modelhub_state/submission_exclusions.jsonl` and are not retried. + ## Concurrent Agents The token pool keeps a local reservation for every in-flight submission, so a @@ -95,7 +102,8 @@ is retained and the runner immediately draws replacement candidates from the same scan instead of retrying the duplicate every cycle. Startup logs and the health response expose `agent_version`; version `2026.08.02.3` or newer includes duplicate replacement behavior, while version `2026.08.02.4` adds adaptive -candidate-window expansion and skip-reason reporting. +candidate-window expansion and skip-reason reporting. Version `2026.08.02.5` +adds fail-closed model/GPU prechecks and persistent uniqueness exclusions. ## Deploy diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 4ff4ccad..ca834e09 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -83,6 +83,10 @@ bash run_poll.sh --dry-run 7 days, 30 days, and older history (up to 3,000 models) when the recent pool is exhausted. - Model verification responses are cached across poll cycles for 15 minutes. Local model/GPU failures cool down after 24 hours instead of remaining permanently blocked. +- Community deduplication is model/GPU-specific: another GPU's adaptation does not block the + current GPU. Every actual submission performs a fresh uncached check for its exact GPU. +- If the community lookup is unavailable, submission is deferred. A platform model-uniqueness + rejection permanently excludes only that model/GPU combination from future local retries. - Each model can be submitted at most once per GPU. - Multiple ModelHub tokens are pooled and used to route submissions to the account with available async capacity. - Concurrent submissions reserve account slots locally, and an account-capacity race automatically falls through to another account. @@ -148,6 +152,7 @@ Persistent local scheduler state is written under `.modelhub_state/`: - `gpu_strategy.json`: GPU ranks, generation progress, and 50/30/20 accepted counters - `account_capacity.json`: learned per-account active-task limits +- `submission_exclusions.jsonl`: non-retryable model/GPU uniqueness rejections ## Verification diff --git a/modelhub_submmit_api/daily_runner.py b/modelhub_submmit_api/daily_runner.py index dba8ee8e..77cfba9c 100644 --- a/modelhub_submmit_api/daily_runner.py +++ b/modelhub_submmit_api/daily_runner.py @@ -95,6 +95,11 @@ def build_parser() -> argparse.ArgumentParser: default=os.getenv("MODELHUB_AGENT_CLAIMS_PATH", str(DEFAULT_CLAIMS_PATH)), help=argparse.SUPPRESS, ) + parser.add_argument( + "--submission-exclusions-path", + default=os.getenv("MODELHUB_SUBMISSION_EXCLUSIONS_PATH", ".modelhub_state/submission_exclusions.jsonl"), + help=argparse.SUPPRESS, + ) parser.add_argument("--history-archive-path", default="history/platform_tasks.jsonl", help=argparse.SUPPRESS) parser.add_argument("--history-archive-limit", type=int, default=5000, help=argparse.SUPPRESS) parser.add_argument( @@ -157,6 +162,11 @@ def make_wave_namespace(base_args: argparse.Namespace, wave: WaveSpec) -> argpar ledger_path=base_args.ledger_path, outcomes_path=getattr(base_args, "outcomes_path", "outcomes/submissions.jsonl"), claims_path=getattr(base_args, "claims_path", str(DEFAULT_CLAIMS_PATH)), + submission_exclusions_path=getattr( + base_args, + "submission_exclusions_path", + ".modelhub_state/submission_exclusions.jsonl", + ), history_archive_path=base_args.history_archive_path, history_archive_limit=base_args.history_archive_limit, hf_base_url=base_args.hf_base_url, @@ -245,7 +255,9 @@ def run_daily_batches( f"[daily] wave_done name={wave.name} " f"candidates={summary['candidateCount']} planned={summary['plannedSubmitCount']} " f"submitted={summary['submittedCount']} skipped={summary['skippedCount']} " - f"duplicates={summary.get('duplicateCount', 0)} failed={summary['failedCount']} " + f"duplicates={summary.get('duplicateCount', 0)} " + f"uniqueness_rejected={summary.get('modelGpuUniquenessRejectedCount', 0)} " + f"failed={summary['failedCount']} " f"skip_reasons={skip_reason_text} " f"remaining_before_run={summary['remainingDailyQuotaBeforeRun']}" ) diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 7bfd70f6..3de05137 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -18,10 +18,18 @@ from history_stats import ( load_ledger, update_history_archive, ) -from modelhub_client import DEFAULT_CAPACITY_STATE_PATH, ModelHubAPIError, ModelHubClient, ModelHubClientPool, is_duplicate_submission_error +from modelhub_client import ( + DEFAULT_CAPACITY_STATE_PATH, + ModelHubAPIError, + ModelHubClient, + ModelHubClientPool, + is_duplicate_submission_error, + is_model_uniqueness_error, +) from models import CandidateModel, HFModelSummary, ModelInspection from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, candidate_key, diversify_candidates +from submission_exclusions import DEFAULT_SUBMISSION_EXCLUSIONS_PATH, SubmissionExclusionStore from task_registry import TASK_SPEC_BY_TYPE, all_task_types, choose_framework_for_task, choose_text_generation_framework, pipeline_tags_for_task_types, task_specs_for_model from template_selector import TemplateSelector @@ -94,6 +102,11 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument("--history-archive-path", default="history/platform_tasks.jsonl", help=argparse.SUPPRESS) parser.add_argument("--history-archive-limit", type=int, default=5000, help=argparse.SUPPRESS) + parser.add_argument( + "--submission-exclusions-path", + default=os.getenv("MODELHUB_SUBMISSION_EXCLUSIONS_PATH", str(DEFAULT_SUBMISSION_EXCLUSIONS_PATH)), + help=argparse.SUPPRESS, + ) parser.add_argument( "--gpu-strategy-state-path", default=os.getenv("MODELHUB_GPU_STRATEGY_STATE_PATH", str(DEFAULT_GPU_STRATEGY_PATH)), @@ -270,12 +283,24 @@ def process_model_for_candidates( target_gpus: list[str], allowed_task_types: list[str], outcome_tracker: OutcomeTracker | None = None, + submission_exclusion_store: SubmissionExclusionStore | None = None, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: specs = [spec for spec in task_specs_for_model(model) if spec.task_type in allowed_task_types] if not specs: return [], [{"repoId": model.repo_id, "reason": f"unsupported_pipeline_tag:{model.pipeline_tag or 'unknown'}"}], [] - processed_gpus = modelhub_client.processed_gpus_for_model(model.repo_id) + try: + if hasattr(modelhub_client, "model_submission_precheck"): + precheck = modelhub_client.model_submission_precheck(model.repo_id) + processed_gpus = set(precheck.get("processedGpus") or set()) + else: + processed_gpus = modelhub_client.processed_gpus_for_model(model.repo_id) + except Exception as exc: + return ( + [], + [{"repoId": model.repo_id, "reason": "community_precheck_unavailable_fail_closed"}], + [{"repoId": model.repo_id, "reason": f"community_precheck_error:{exc}"}], + ) candidates: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] failed: list[dict[str, Any]] = [] @@ -283,6 +308,11 @@ def process_model_for_candidates( pending_task_types_by_gpu: list[tuple[str, list[str]]] = [] for target_gpu in target_gpus: + if submission_exclusion_store is not None and submission_exclusion_store.is_blocked(model.repo_id, target_gpu): + skipped.append( + {"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "model_gpu_uniqueness_blocklist"} + ) + continue if target_gpu in processed_gpus: skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "already_processed_for_gpu"}) continue @@ -391,6 +421,7 @@ def collect_candidates_from_models( target_gpus: list[str], selected_task_types: list[str], outcome_tracker: OutcomeTracker | None, + submission_exclusion_store: SubmissionExclusionStore | None, read_concurrency: int, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], int]: candidates: list[dict[str, Any]] = [] @@ -419,6 +450,7 @@ def collect_candidates_from_models( target_gpus=target_gpus, allowed_task_types=selected_task_types, outcome_tracker=outcome_tracker, + submission_exclusion_store=submission_exclusion_store, ): index for index, model in enumerate(chunk) } @@ -445,6 +477,27 @@ def submit_candidate( candidate: dict[str, Any], modelhub_client: ModelHubClient, ) -> dict[str, Any]: + if hasattr(modelhub_client, "model_submission_precheck"): + try: + precheck = modelhub_client.model_submission_precheck(candidate["repoId"], force_refresh=True) + except Exception as exc: + print( + f"[submit] deferred repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=community_precheck_unavailable", + flush=True, + ) + return { + "outcome": "precheck_deferred", + "candidate": candidate, + "reason": f"community_precheck_error:{exc}", + } + if candidate["targetGpu"] in set(precheck.get("processedGpus") or set()): + return { + "outcome": "duplicate", + "candidate": candidate, + "reason": "already_processed_for_gpu", + } + payload = { "modelAddress": candidate["modelAddress"], "taskType": candidate["taskType"], @@ -469,6 +522,17 @@ def submit_candidate( "responseData": response.get("data"), } except ModelHubAPIError as exc: + if is_model_uniqueness_error(exc): + print( + f"[submit] skipped repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=model_uniqueness_rejected", + flush=True, + ) + return { + "outcome": "uniqueness_rejected", + "candidate": candidate, + "reason": str(exc), + } if is_duplicate_submission_error(exc): print( f"[submit] skipped repo={candidate['repoId']} gpu={candidate['targetGpu']} " @@ -505,6 +569,18 @@ def make_run_dir(runs_dir: Path, now) -> Path: raise RuntimeError(f"Unable to allocate a unique run directory under {runs_dir}") +def one_candidate_per_model(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: + selected: list[dict[str, Any]] = [] + seen_models: set[str] = set() + for candidate in candidates: + model_id = str(candidate.get("repoId") or candidate.get("modelAddress") or "") + if not model_id or model_id in seen_models: + continue + seen_models.add(model_id) + selected.append(candidate) + return selected + + def run_submission( args: argparse.Namespace, *, @@ -543,6 +619,9 @@ def run_submission( history_archive_path.parent.mkdir(parents=True, exist_ok=True) outcome_tracker = outcome_tracker or OutcomeTracker(Path(args.outcomes_path)) + submission_exclusion_store = SubmissionExclusionStore( + Path(getattr(args, "submission_exclusions_path", DEFAULT_SUBMISSION_EXCLUSIONS_PATH)) + ) strategy_manager: GPUStrategyManager | None = None strategy_summary: dict[str, Any] = {"enabled": False} strategy_enabled = ( @@ -638,9 +717,12 @@ def run_submission( "plannedSubmitCount": 0, "submittedCount": 0, "duplicateCount": 0, + "modelGpuUniquenessRejectedCount": 0, "skippedCount": 0, "skipReasonCounts": {}, "failedCount": 0, + "failureReasonCounts": {}, + "submissionExclusionsPath": str(submission_exclusion_store.path), "warnings": report.get("warnings", []), "runDir": str(run_dir), "outcomeSyncCount": synced_count, @@ -740,6 +822,7 @@ def run_submission( target_gpus=target_gpus, selected_task_types=selected_task_types, outcome_tracker=outcome_tracker, + submission_exclusion_store=submission_exclusion_store, read_concurrency=max(1, args.read_concurrency), ) candidates.extend(stage_candidates) @@ -767,6 +850,7 @@ def run_submission( submitted: list[dict[str, Any]] = [] duplicate_candidates: list[dict[str, Any]] = [] + uniqueness_rejected_candidates: list[dict[str, Any]] = [] target_submit_count = resolve_max_submit_count( args=args, planned_count=len(candidates), @@ -806,7 +890,9 @@ def run_submission( candidate for candidate in diversified_candidates if candidate_key(candidate) not in attempted_keys + and not submission_exclusion_store.is_blocked(candidate["repoId"], candidate["targetGpu"]) ] + remaining_candidates = one_candidate_per_model(remaining_candidates) desired_count = min( target_submit_count - len(submitted), max_submit_attempts - len(attempted_candidates), @@ -843,12 +929,30 @@ def run_submission( batch_submitted_candidates: list[dict[str, Any]] = [] batch_duplicate_candidates: list[dict[str, Any]] = [] + batch_uniqueness_rejected_candidates: list[dict[str, Any]] = [] batch_failed_candidates: list[dict[str, Any]] = [] for index in range(len(batch_candidates)): result = ordered_results.get(index) if result is None: continue candidate = result["candidate"] + if result["outcome"] == "uniqueness_rejected": + duplicate_candidates.append(candidate) + uniqueness_rejected_candidates.append(candidate) + batch_uniqueness_rejected_candidates.append(candidate) + submission_exclusion_store.block( + candidate["repoId"], + candidate["targetGpu"], + reason=result.get("reason", "model_uniqueness_rejected"), + ) + skipped.append( + { + "repoId": candidate["repoId"], + "targetGpu": candidate["targetGpu"], + "reason": "model_uniqueness_rejected", + } + ) + continue if result["outcome"] == "duplicate": duplicate_candidates.append(candidate) batch_duplicate_candidates.append(candidate) @@ -860,7 +964,7 @@ def run_submission( } ) continue - if result["outcome"] == "failed": + if result["outcome"] in {"failed", "precheck_deferred"}: batch_failed_candidates.append(candidate) failed.append( { @@ -903,7 +1007,9 @@ def run_submission( submit_time=result["submitTime"], ) - claim_store.mark_submitted([*batch_submitted_candidates, *batch_duplicate_candidates]) + claim_store.mark_submitted( + [*batch_submitted_candidates, *batch_duplicate_candidates, *batch_uniqueness_rejected_candidates] + ) claim_store.release(batch_failed_candidates) if strategy_manager is not None: strategy_manager.record_accepted(batch_submitted_candidates) @@ -950,11 +1056,13 @@ def run_submission( ), "plannedSubmitCount": len(attempted_candidates), "duplicateCount": len(duplicate_candidates), + "modelGpuUniquenessRejectedCount": len(uniqueness_rejected_candidates), "submittedCount": len(submitted), "skippedCount": len(skipped), "skipReasonCounts": dict(skip_reason_counts.most_common()), "failedCount": len(failed), "failureReasonCounts": dict(failure_reason_counts.most_common()), + "submissionExclusionsPath": str(submission_exclusion_store.path), "submitConcurrencyUsed": submit_workers, "warnings": report.get("warnings", []), "runDir": str(run_dir), diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 711de90d..d6ea3242 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -21,6 +21,23 @@ class ModelHubAPIError(RuntimeError): self.payload = payload +def parse_model_submission_precheck(payload: Any) -> dict[str, Any]: + """Validate the community lookup response instead of failing open.""" + if not isinstance(payload, dict): + raise ModelHubAPIError("Community model precheck returned a non-object response", payload=payload) + data = payload.get("data") + if not isinstance(data, dict) or "verifyResult" not in data: + raise ModelHubAPIError("Community model precheck response is incomplete", payload=payload) + verify_result = data.get("verifyResult") or {} + if not isinstance(verify_result, dict): + raise ModelHubAPIError("Community model precheck verifyResult is invalid", payload=payload) + + return { + "isInDB": data.get("isInDB") is True, + "processedGpus": set(str(gpu) for gpu in verify_result), + } + + class ModelHubClient: def __init__( self, @@ -42,9 +59,13 @@ class ModelHubClient: retries=retries, ) - def search_by_model_id(self, model_id: str) -> dict[str, Any]: + def search_by_model_id(self, model_id: str, *, force_refresh: bool = False) -> dict[str, Any]: + del force_refresh return self._request("GET", "/api/computility/models/search-by-model-id", query={"modelId": model_id}) + def model_submission_precheck(self, model_id: str, *, force_refresh: bool = False) -> dict[str, Any]: + return parse_model_submission_precheck(self.search_by_model_id(model_id, force_refresh=force_refresh)) + def is_model_processed_for_gpu(self, model_id: str, target_gpu: str) -> bool: gpu_result = self.get_verify_result_map(model_id).get(target_gpu) if gpu_result is None: @@ -56,7 +77,7 @@ class ModelHubClient: return ((payload.get("data") or {}).get("verifyResult") or {}) def processed_gpus_for_model(self, model_id: str) -> set[str]: - return set(self.get_verify_result_map(model_id).keys()) + return set(self.model_submission_precheck(model_id)["processedGpus"]) def list_tasks_page( self, @@ -256,6 +277,16 @@ DUPLICATE_SUBMISSION_MARKERS = ( "already in progress", ) +MODEL_UNIQUENESS_ERROR_MARKERS = ( + "模型唯一性检查", + "唯一性检查没有通过", + "模型已存在", + "model uniqueness", + "uniqueness check", + "duplicate model", + "model already exists", +) + DEFAULT_CAPACITY_STATE_PATH = Path(".modelhub_state/account_capacity.json") @@ -275,6 +306,13 @@ def is_duplicate_submission_error(error: ModelHubAPIError) -> bool: return any(marker in message for marker in DUPLICATE_SUBMISSION_MARKERS) +def is_model_uniqueness_error(error: ModelHubAPIError) -> bool: + message = str(error).strip().lower() + if isinstance(error.payload, dict): + message = f"{message} {error.payload.get('message') or ''}".lower() + return any(marker in message for marker in MODEL_UNIQUENESS_ERROR_MARKERS) + + class ModelHubClientPool: def __init__( self, @@ -385,13 +423,6 @@ class ModelHubClientPool: except Exception: return max(0, max_count - 1) - def _safe_search_by_model_id(self, client: ModelHubClient, model_id: str) -> dict[str, Any]: - try: - payload = client.search_by_model_id(model_id) - return payload if isinstance(payload, dict) else {} - except Exception: - return {} - def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]: try: return client.list_tasks(**kwargs) @@ -546,29 +577,33 @@ class ModelHubClientPool: if now - cached_at >= self._verify_cache_ttl: self._verify_cache.pop(model_id, None) - def search_by_model_id(self, model_id: str) -> dict[str, Any]: + def search_by_model_id(self, model_id: str, *, force_refresh: bool = False) -> dict[str, Any]: # Reuse recent model verification results across short poll cycles. - with self._state_lock: - cached = self._read_from_cache(model_id) - if cached is not None: - return cached + if not force_refresh: + with self._state_lock: + cached = self._read_from_cache(model_id) + if cached is not None: + return cached # Only query ONE client (the reader) instead of fanning out to all clients. # verifyResult is model-specific platform data, not account-specific. - response = self._safe_search_by_model_id(self._reader, model_id) + response = self._reader.search_by_model_id(model_id) if not isinstance(response, dict): - response = {"code": 0, "data": {"verifyResult": {}}} + raise ModelHubAPIError("Community model precheck returned a non-object response", payload=response) with self._state_lock: self._write_to_cache(model_id, response) return response + def model_submission_precheck(self, model_id: str, *, force_refresh: bool = False) -> dict[str, Any]: + return parse_model_submission_precheck(self.search_by_model_id(model_id, force_refresh=force_refresh)) + def get_verify_result_map(self, model_id: str) -> dict[str, Any]: payload = self.search_by_model_id(model_id) return ((payload.get("data") or {}).get("verifyResult") or {}) def processed_gpus_for_model(self, model_id: str) -> set[str]: - return set(self.get_verify_result_map(model_id).keys()) + return set(self.model_submission_precheck(model_id)["processedGpus"]) def _reserve_account(self, excluded: set[int]) -> tuple[int, int] | None: with self._state_lock: diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index b1dd5ff5..15cacf3f 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -74,6 +74,11 @@ def build_parser() -> argparse.ArgumentParser: default=os.getenv("MODELHUB_AGENT_CLAIMS_PATH", str(DEFAULT_CLAIMS_PATH)), help=argparse.SUPPRESS, ) + parser.add_argument( + "--submission-exclusions-path", + default=os.getenv("MODELHUB_SUBMISSION_EXCLUSIONS_PATH", ".modelhub_state/submission_exclusions.jsonl"), + help=argparse.SUPPRESS, + ) parser.add_argument("--history-archive-path", default="history/platform_tasks.jsonl", help=argparse.SUPPRESS) parser.add_argument("--history-archive-limit", type=int, default=5000, help=argparse.SUPPRESS) parser.add_argument( diff --git a/modelhub_submmit_api/submission_exclusions.py b/modelhub_submmit_api/submission_exclusions.py new file mode 100644 index 00000000..9fd8e482 --- /dev/null +++ b/modelhub_submmit_api/submission_exclusions.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import threading +from pathlib import Path +from typing import Any + +from common import read_jsonl, update_jsonl, utc_now + + +DEFAULT_SUBMISSION_EXCLUSIONS_PATH = Path(".modelhub_state/submission_exclusions.jsonl") + + +def exclusion_key(model_id: str, target_gpu: str) -> str: + return f"{model_id}|{target_gpu}" + + +class SubmissionExclusionStore: + """Persistent model/GPU exclusions for non-retryable platform rejections.""" + + def __init__(self, path: Path | str = DEFAULT_SUBMISSION_EXCLUSIONS_PATH) -> None: + self.path = Path(path) + self._lock = threading.Lock() + self._blocked = { + exclusion_key(str(record.get("modelId")), str(record.get("targetGpu"))) + for record in read_jsonl(self.path) + if record.get("modelId") and record.get("targetGpu") + } + + def is_blocked(self, model_id: str, target_gpu: str) -> bool: + with self._lock: + return exclusion_key(model_id, target_gpu) in self._blocked + + def block(self, model_id: str, target_gpu: str, *, reason: str) -> None: + if not model_id or not target_gpu: + return + key = exclusion_key(model_id, target_gpu) + now = utc_now().isoformat() + + def update(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + existing = next( + ( + record + for record in records + if exclusion_key(str(record.get("modelId")), str(record.get("targetGpu"))) == key + ), + None, + ) + if existing is None: + records.append( + { + "modelId": model_id, + "targetGpu": target_gpu, + "reason": reason, + "firstSeenAt": now, + "lastSeenAt": now, + "occurrences": 1, + } + ) + else: + existing["reason"] = reason + existing["lastSeenAt"] = now + existing["occurrences"] = max(1, int(existing.get("occurrences") or 1)) + 1 + return records + + update_jsonl(self.path, update) + with self._lock: + self._blocked.add(key) diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index d166abb3..3ae78e79 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.02.4" +AGENT_VERSION = "2026.08.02.5" diff --git a/tests/test_submission_safety.py b/tests/test_submission_safety.py new file mode 100644 index 00000000..ac0cf340 --- /dev/null +++ b/tests/test_submission_safety.py @@ -0,0 +1,225 @@ +from __future__ import annotations + +import sys +import tempfile +import unittest +from datetime import datetime, timezone +from pathlib import Path + + +PACKAGE_DIR = Path(__file__).resolve().parents[1] / "modelhub_submmit_api" +if str(PACKAGE_DIR) in sys.path: + sys.path.remove(str(PACKAGE_DIR)) +sys.path.insert(0, str(PACKAGE_DIR)) + +from main import build_parser, one_candidate_per_model, process_model_for_candidates, run_submission, submit_candidate # noqa: E402 +from modelhub_client import ( # noqa: E402 + ModelHubAPIError, + ModelHubClientPool, + is_model_uniqueness_error, + parse_model_submission_precheck, +) +from models import HFModelSummary, ModelInspection # noqa: E402 +from submission_exclusions import SubmissionExclusionStore # noqa: E402 +from template_selector import TemplateSelector # noqa: E402 + + +class SafeDiscovery: + def __init__(self) -> None: + self.model = HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + + def list_recent_models(self, **_kwargs) -> list[HFModelSummary]: # noqa: ANN003 + return [self.model] + + @staticmethod + def inspect_model(model: HFModelSummary) -> ModelInspection: + return ModelInspection(repo_id=model.repo_id, weight_files=["model.safetensors"]) + + +class OtherGpuProcessedClient: + @staticmethod + def model_submission_precheck(_model_id: str, *, force_refresh: bool = False) -> dict: + del force_refresh + return {"processedGpus": {"Biren_166m"}, "isInDB": True} + + +class FailingLookupClient: + @staticmethod + def search_by_model_id(_model_id: str) -> dict: + raise ModelHubAPIError("temporary lookup outage") + + +class ExactGpuProcessedClient: + def __init__(self) -> None: + self.add_calls = 0 + + @staticmethod + def model_submission_precheck(_model_id: str, *, force_refresh: bool = False) -> dict: + del force_refresh + return {"processedGpus": {"Vastai_va16"}, "isInDB": True} + + def add_task(self, _payload: dict) -> dict: + self.add_calls += 1 + return {"code": 0, "data": {"id": "should-not-submit"}} + + +class UniquenessRejectingClient: + @staticmethod + def add_task(_payload: dict) -> dict: + raise ModelHubAPIError("模型唯一性检查没有通过，无法进行同步") + + +class UniquenessRunClient(UniquenessRejectingClient): + def __init__(self) -> None: + self.add_calls = 0 + + @staticmethod + def available_submit_slots() -> int: + return 1 + + @staticmethod + def list_tasks_page(**_kwargs) -> dict: # noqa: ANN003 + return {"code": 0, "data": {"records": []}} + + @staticmethod + def processed_gpus_for_model(_model_id: str) -> set[str]: + return set() + + def add_task(self, _payload: dict) -> dict: + self.add_calls += 1 + return super().add_task(_payload) + + +def candidate(target_gpu: str = "Vastai_va16") -> dict: + return { + "repoId": "owner/model", + "modelAddress": "https://modelscope.cn/models/owner/model", + "taskType": "text-generation", + "targetGpu": target_gpu, + "framework": "vllm", + "configParams": "framework: vllm", + } + + +class SubmissionSafetyTests(unittest.TestCase): + def test_precheck_tracks_processed_gpus_without_blocking_other_gpus(self) -> None: + payload = { + "code": 0, + "data": { + "isInDB": True, + "verifyResult": { + "Biren_166m": { + "result": "已验证", + "records": [{"verifyResult": 1}], + } + }, + }, + } + precheck = parse_model_submission_precheck(payload) + self.assertEqual({"Biren_166m"}, precheck["processedGpus"]) + + model = HFModelSummary( + repo_id="owner/model", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + candidates, skipped, failed = process_model_for_candidates( + model=model, + hf_discovery=SafeDiscovery(), # type: ignore[arg-type] + modelhub_client=OtherGpuProcessedClient(), # type: ignore[arg-type] + template_selector=TemplateSelector(), + target_gpus=["Biren_166m", "Vastai_va16"], + allowed_task_types=["text-generation"], + ) + self.assertEqual(["Vastai_va16"], [item["targetGpu"] for item in candidates]) + self.assertEqual("already_processed_for_gpu", skipped[0]["reason"]) + self.assertEqual([], failed) + + def test_lookup_failure_is_fail_closed(self) -> None: + pool = ModelHubClientPool([FailingLookupClient()]) # type: ignore[arg-type] + with self.assertRaises(ModelHubAPIError): + pool.model_submission_precheck("owner/model") + + def test_submit_precheck_stops_an_exact_processed_gpu(self) -> None: + client = ExactGpuProcessedClient() + result = submit_candidate(candidate(), client) # type: ignore[arg-type] + self.assertEqual("duplicate", result["outcome"]) + self.assertEqual(0, client.add_calls) + + def test_uniqueness_rejection_is_non_retryable_for_that_combination(self) -> None: + error = ModelHubAPIError("模型唯一性检查没有通过，无法进行同步") + self.assertTrue(is_model_uniqueness_error(error)) + result = submit_candidate(candidate(), UniquenessRejectingClient()) # type: ignore[arg-type] + self.assertEqual("uniqueness_rejected", result["outcome"]) + + with tempfile.TemporaryDirectory() as temporary_dir: + path = Path(temporary_dir) / "exclusions.jsonl" + store = SubmissionExclusionStore(path) + store.block("owner/model", "Vastai_va16", reason=str(error)) + reloaded = SubmissionExclusionStore(path) + self.assertTrue(reloaded.is_blocked("owner/model", "Vastai_va16")) + self.assertFalse(reloaded.is_blocked("owner/model", "Biren_166m")) + + def test_run_persists_uniqueness_rejection_and_does_not_retry_it(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + args = build_parser().parse_args( + [ + "--gpus", + "Vastai_va16", + "--task-types", + "text-generation", + "--limit", + "1", + "--max-scan-models", + "1", + "--skip-outcome-sync", + "--skip-history-archive", + ] + ) + args.runs_dir = str(root / "runs") + args.ledger_path = str(root / "ledger.jsonl") + args.outcomes_path = str(root / "outcomes.jsonl") + args.claims_path = str(root / "claims.jsonl") + args.submission_exclusions_path = str(root / "exclusions.jsonl") + args.history_archive_path = str(root / "history.jsonl") + client = UniquenessRunClient() + discovery = SafeDiscovery() + + first = run_submission( + args, + now=datetime(2026, 1, 1, 12, tzinfo=timezone.utc), + hf_discovery=discovery, # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + ) + second = run_submission( + args, + now=datetime(2026, 1, 1, 12, 1, tzinfo=timezone.utc), + hf_discovery=discovery, # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + ) + + self.assertEqual(1, first["modelGpuUniquenessRejectedCount"]) + self.assertEqual(1, client.add_calls) + self.assertEqual(0, second["candidateCount"]) + self.assertEqual(1, second["skipReasonCounts"]["model_gpu_uniqueness_blocklist"]) + + def test_concurrent_batch_uses_at_most_one_gpu_per_model(self) -> None: + selected = one_candidate_per_model( + [ + candidate("Vastai_va16"), + candidate("Biren_166m"), + {**candidate("Biren_166m"), "repoId": "owner/other"}, + ] + ) + self.assertEqual(["owner/model", "owner/other"], [item["repoId"] for item in selected]) + + +if __name__ == "__main__": + unittest.main()