diff --git a/README.md b/README.md index 1fae39f7..f5ced5da 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,12 @@ different candidate ordering (derived from `STRATEGY_ID`, instance ID, or hostname), which reduces duplicate work while the platform remains the final authority for account capacity and model/GPU uniqueness. +If the platform reports that a model/GPU is already being validated, the claim +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.1` or newer includes +this behavior. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". diff --git a/main.py b/main.py index 5610a6de..c4103ee8 100644 --- a/main.py +++ b/main.py @@ -10,6 +10,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from modelhub_submmit_api.defaults import EMBEDDED_MODELHUB_XC_TOKEN, EMBEDDED_MODELSCOPE_TOKEN +from modelhub_submmit_api.version import AGENT_VERSION HOST = "0.0.0.0" @@ -74,6 +75,7 @@ def _worker_command() -> list[str]: def _config() -> dict[str, object]: return { + "agent_version": AGENT_VERSION, "strategy_id_present": bool(os.getenv("STRATEGY_ID")), "modelscope_token_present": bool(os.getenv("MODELSCOPE_API_TOKEN") or os.getenv("MODELSCOPE_TOKEN") or EMBEDDED_MODELSCOPE_TOKEN), "modelhub_auth_present": _has_modelhub_auth(), @@ -92,7 +94,7 @@ class Handler(BaseHTTPRequestHandler): if worker is not None and worker.poll() is not None: self._send_json({"status": "worker_exited", "config": _config()}, status=500) return - self._send_json({"status": "ok"}) + self._send_json({"status": "ok", "config": _config()}) return if self.path == "/": @@ -135,6 +137,7 @@ def _handle_signal(signum: int, _frame: object) -> None: def main() -> int: global config_error, worker + print(f"modelhub-submmit-agent version={AGENT_VERSION}", flush=True) signal.signal(signal.SIGTERM, _handle_signal) signal.signal(signal.SIGINT, _handle_signal) diff --git a/modelhub_submmit_api/daily_runner.py b/modelhub_submmit_api/daily_runner.py index 2ac970e1..278b10f9 100644 --- a/modelhub_submmit_api/daily_runner.py +++ b/modelhub_submmit_api/daily_runner.py @@ -198,7 +198,8 @@ 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"failed={summary['failedCount']} remaining_before_run={summary['remainingDailyQuotaBeforeRun']}" + f"duplicates={summary.get('duplicateCount', 0)} failed={summary['failedCount']} " + f"remaining_before_run={summary['remainingDailyQuotaBeforeRun']}" ) if summary.get("platformAvailableSlotsBeforeRun") == 0: @@ -267,6 +268,14 @@ def finalize_daily_run( stopped_reason: str, ) -> dict[str, Any]: last_wave_summary = wave_results[-1]["summary"] if wave_results else {} + duplicate_total = sum( + int(wave_result.get("summary", {}).get("duplicateCount", 0) or 0) + for wave_result in wave_results + ) + failed_total = sum( + int(wave_result.get("summary", {}).get("failedCount", 0) or 0) + for wave_result in wave_results + ) summary = { "generatedAt": now.isoformat(), "dryRun": bool(base_args.dry_run), @@ -275,6 +284,8 @@ def finalize_daily_run( "rounds": base_args.rounds, "attemptedWaves": attempted_waves, "submittedTotal": submitted_total, + "duplicateTotal": duplicate_total, + "failedTotal": failed_total, "stoppedReason": stopped_reason, "dailyRunDir": str(daily_run_dir), "remainingDailyQuotaBeforeRun": last_wave_summary.get("remainingDailyQuotaBeforeRun"), diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 9f2850f1..c34059fe 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -16,10 +16,10 @@ from history_stats import ( load_ledger, update_history_archive, ) -from modelhub_client import ModelHubAPIError, ModelHubClient, ModelHubClientPool +from modelhub_client import ModelHubAPIError, ModelHubClient, ModelHubClientPool, is_duplicate_submission_error from models import CandidateModel, HFModelSummary, ModelInspection from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker -from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, diversify_candidates +from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, candidate_key, diversify_candidates 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 @@ -319,6 +319,17 @@ def submit_candidate( "responseData": response.get("data"), } except ModelHubAPIError as exc: + if is_duplicate_submission_error(exc): + print( + f"[submit] skipped repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=already_validating", + flush=True, + ) + return { + "outcome": "duplicate", + "candidate": candidate, + "reason": str(exc), + } print( f"[submit] failed repo={candidate['repoId']} gpu={candidate['targetGpu']} " f"framework={candidate['framework']} reason={exc}", @@ -458,8 +469,11 @@ def run_submission( "scanLimit": 0, "scannedModels": 0, "candidateCount": 0, + "targetSubmitCount": 0, + "maxSubmitAttempts": 0, "plannedSubmitCount": 0, "submittedCount": 0, + "duplicateCount": 0, "skippedCount": 0, "failedCount": 0, "warnings": report.get("warnings", []), @@ -550,7 +564,8 @@ def run_submission( write_jsonl(run_dir / "candidates.jsonl", candidates) submitted: list[dict[str, Any]] = [] - planned_submit_count = resolve_max_submit_count( + duplicate_candidates: list[dict[str, Any]] = [] + target_submit_count = resolve_max_submit_count( args=args, planned_count=len(candidates), remaining_daily_quota=remaining_daily_quota, @@ -558,94 +573,134 @@ def run_submission( instance_id = runtime_instance_id() diversified_candidates = diversify_candidates(candidates, instance_id=instance_id) claim_store: SubmissionClaimStore | None = None + attempted_candidates: list[dict[str, Any]] = [] + submit_workers = 1 + if args.dry_run: - planned_candidates = diversified_candidates[:planned_submit_count] + attempted_candidates = diversified_candidates[:target_submit_count] else: claim_store = SubmissionClaimStore( Path(getattr(args, "claims_path", DEFAULT_CLAIMS_PATH)), owner_id=instance_id, ) - planned_candidates = claim_store.claim(diversified_candidates, limit=planned_submit_count) - submit_workers = 1 - if not args.dry_run: - submit_workers = resolve_submit_concurrency( - args, - modelhub_client=modelhub_client, - planned_submit_count=len(planned_candidates), + attempted_keys: set[str] = set() + attempt_multiplier = max(1, int(getattr(args, "scan_multiplier", 4) or 1)) + max_submit_attempts = min( + len(diversified_candidates), + max(target_submit_count, target_submit_count * attempt_multiplier), ) - with ThreadPoolExecutor(max_workers=submit_workers) as executor: - futures = { - executor.submit( - submit_candidate, - candidate, - modelhub_client, - ): index - for index, candidate in enumerate(planned_candidates) - } - ordered_results: dict[int, dict[str, Any]] = {} - for future in as_completed(futures): - index = futures[future] - try: - ordered_results[index] = future.result() - except Exception as exc: - candidate = planned_candidates[index] - ordered_results[index] = { - "outcome": "failed", - "candidate": candidate, - "reason": str(exc), - } - for index in range(len(planned_candidates)): - result = ordered_results.get(index) - if result is None: - continue - candidate = result["candidate"] - if result["outcome"] == "failed": - failed.append( + # A batch can contain candidates another machine has already submitted. + # Keep those duplicate claims and immediately draw replacements from the + # already-scanned pool until the desired number of real submissions is + # reached or account capacity is genuinely exhausted. + while len(submitted) < target_submit_count and len(attempted_candidates) < max_submit_attempts: + remaining_candidates = [ + candidate + for candidate in diversified_candidates + if candidate_key(candidate) not in attempted_keys + ] + desired_count = min( + target_submit_count - len(submitted), + max_submit_attempts - len(attempted_candidates), + ) + batch_candidates = claim_store.claim(remaining_candidates, limit=desired_count) + if not batch_candidates: + break + attempted_candidates.extend(batch_candidates) + attempted_keys.update(candidate_key(candidate) for candidate in batch_candidates) + + batch_workers = resolve_submit_concurrency( + args, + modelhub_client=modelhub_client, + planned_submit_count=len(batch_candidates), + ) + submit_workers = max(submit_workers, batch_workers) + with ThreadPoolExecutor(max_workers=batch_workers) as executor: + futures = { + executor.submit(submit_candidate, candidate, modelhub_client): index + for index, candidate in enumerate(batch_candidates) + } + ordered_results: dict[int, dict[str, Any]] = {} + for future in as_completed(futures): + index = futures[future] + try: + ordered_results[index] = future.result() + except Exception as exc: + candidate = batch_candidates[index] + ordered_results[index] = { + "outcome": "failed", + "candidate": candidate, + "reason": str(exc), + } + + batch_submitted_candidates: list[dict[str, Any]] = [] + batch_duplicate_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"] == "duplicate": + duplicate_candidates.append(candidate) + batch_duplicate_candidates.append(candidate) + skipped.append( + { + "repoId": candidate["repoId"], + "targetGpu": candidate["targetGpu"], + "reason": "already_validating_on_platform", + } + ) + continue + if result["outcome"] == "failed": + batch_failed_candidates.append(candidate) + failed.append( + { + "repoId": candidate["repoId"], + "targetGpu": candidate["targetGpu"], + "framework": candidate["framework"], + "taskType": candidate["taskType"], + "reason": result.get("reason", "submission_failed"), + } + ) + continue + + batch_submitted_candidates.append(candidate) + submitted_record = { + **candidate, + "submitTime": result["submitTime"], + "taskId": result["taskId"], + "responseData": result["responseData"], + } + submitted.append(submitted_record) + append_ledger_entry( + ledger_path, { - "repoId": candidate["repoId"], + "modelId": candidate["repoId"], + "modelAddress": candidate["modelAddress"], "targetGpu": candidate["targetGpu"], "framework": candidate["framework"], + "templateId": candidate["templateId"], + "taskId": result["taskId"], "taskType": candidate["taskType"], - "reason": result.get("reason", "submission_failed"), - } + "submitTime": result["submitTime"], + }, + ) + outcome_tracker.record_submission( + model_id=candidate["repoId"], + target_gpu=candidate["targetGpu"], + framework=candidate["framework"], + task_type=candidate["taskType"], + task_id=result["taskId"], + submit_time=result["submitTime"], ) - continue - submitted_record = { - **candidate, - "submitTime": result["submitTime"], - "taskId": result["taskId"], - "responseData": result["responseData"], - } - submitted.append(submitted_record) - append_ledger_entry( - ledger_path, - { - "modelId": candidate["repoId"], - "modelAddress": candidate["modelAddress"], - "targetGpu": candidate["targetGpu"], - "framework": candidate["framework"], - "templateId": candidate["templateId"], - "taskId": result["taskId"], - "taskType": candidate["taskType"], - "submitTime": result["submitTime"], - }, - ) - outcome_tracker.record_submission( - model_id=candidate["repoId"], - target_gpu=candidate["targetGpu"], - framework=candidate["framework"], - task_type=candidate["taskType"], - task_id=result["taskId"], - submit_time=result["submitTime"], - ) + claim_store.mark_submitted([*batch_submitted_candidates, *batch_duplicate_candidates]) + claim_store.release(batch_failed_candidates) - if claim_store is not None: - submitted_candidates = [result["candidate"] for result in ordered_results.values() if result.get("outcome") == "submitted"] - failed_candidates = [result["candidate"] for result in ordered_results.values() if result.get("outcome") != "submitted"] - claim_store.mark_submitted(submitted_candidates) - claim_store.release(failed_candidates) + if hasattr(modelhub_client, "available_submit_slots") and modelhub_client.available_submit_slots() <= 0: + break write_jsonl(run_dir / "submitted.jsonl", submitted) write_jsonl(run_dir / "skipped.jsonl", skipped) @@ -671,7 +726,13 @@ def run_submission( "scanLimit": scan_limit, "scannedModels": len(models), "candidateCount": len(candidates), - "plannedSubmitCount": len(planned_candidates), + "targetSubmitCount": target_submit_count, + "maxSubmitAttempts": 0 if args.dry_run else min( + len(diversified_candidates), + max(target_submit_count, target_submit_count * max(1, int(getattr(args, "scan_multiplier", 4) or 1))), + ), + "plannedSubmitCount": len(attempted_candidates), + "duplicateCount": len(duplicate_candidates), "submittedCount": len(submitted), "skippedCount": len(skipped), "failedCount": len(failed), diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 45fc4190..5bb287b4 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -246,6 +246,15 @@ CAPACITY_ERROR_MARKERS = ( "active task limit", ) +DUPLICATE_SUBMISSION_MARKERS = ( + "正在验证中", + "请勿重复提交", + "重复提交", + "already validating", + "already being validated", + "already in progress", +) + def is_capacity_error(error: ModelHubAPIError) -> bool: if error.code in {409, 429}: @@ -256,6 +265,13 @@ def is_capacity_error(error: ModelHubAPIError) -> bool: return any(marker in message for marker in CAPACITY_ERROR_MARKERS) +def is_duplicate_submission_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 DUPLICATE_SUBMISSION_MARKERS) + + class ModelHubClientPool: def __init__( self, diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index a772645d..a7fa62db 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -17,6 +17,7 @@ from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker from runner_common import DEFAULT_KEY_PATH, ensure_tokens from submission_claims import DEFAULT_CLAIMS_PATH from template_selector import TemplateSelector +from version import AGENT_VERSION DEFAULT_POLL_RUNS_DIR = Path("poll_runs") @@ -122,7 +123,7 @@ def run_poll_loop( OUTCOME_SYNC_INTERVAL = 3 STATS_PRINT_INTERVAL = 10 - log(f"[poll] poll_run_dir={poll_run_dir}") + log(f"[poll] version={AGENT_VERSION} poll_run_dir={poll_run_dir}") log( f"[poll] target={base_args.daily_target} dry_run={str(bool(base_args.dry_run)).lower()} " f"poll_interval={base_args.poll_interval_seconds}s idle_interval={base_args.idle_interval_seconds}s" @@ -165,6 +166,7 @@ def run_poll_loop( log( f"[poll] cycle_done submitted_total={cycle_summary['submittedTotal']} " + f"duplicates={cycle_summary.get('duplicateTotal', 0)} " f"remaining_before_run={remaining_before_run if remaining_before_run is not None else 'n/a'} " f"stop={cycle_summary['stoppedReason']}" ) @@ -174,8 +176,14 @@ def run_poll_loop( break if cycle_summary["submittedTotal"] <= 0: - log(f"[poll] cycle={cycles} sleep={base_args.idle_interval_seconds}s reason=no_new_submissions") - time.sleep(base_args.idle_interval_seconds) + duplicate_total = int(cycle_summary.get("duplicateTotal", 0) or 0) + if duplicate_total > 0 and (available_slots is None or available_slots > 0): + retry_delay = max(1, int(getattr(base_args, "post_cycle_cooldown_seconds", 2) or 2)) + log(f"[poll] cycle={cycles} sleep={retry_delay}s reason=duplicates_need_replacement_candidates") + time.sleep(retry_delay) + else: + log(f"[poll] cycle={cycles} sleep={base_args.idle_interval_seconds}s reason=no_new_submissions") + time.sleep(base_args.idle_interval_seconds) continue if cycles % OUTCOME_SYNC_INTERVAL == 0: diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py new file mode 100644 index 00000000..06caf026 --- /dev/null +++ b/modelhub_submmit_api/version.py @@ -0,0 +1 @@ +AGENT_VERSION = "2026.08.02.1" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 3461cc12..0fa5e634 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -15,8 +15,9 @@ if str(PACKAGE_DIR) in sys.path: sys.path.insert(0, str(PACKAGE_DIR)) from common import read_jsonl # noqa: E402 -from main import make_run_dir # noqa: E402 -from modelhub_client import ModelHubAPIError, ModelHubClientPool # noqa: E402 +from main import build_parser, make_run_dir, run_submission, submit_candidate # noqa: E402 +from modelhub_client import ModelHubAPIError, ModelHubClientPool, is_duplicate_submission_error # noqa: E402 +from models import HFModelSummary, ModelInspection # noqa: E402 from outcome_tracker import OutcomeTracker # noqa: E402 from submission_claims import SubmissionClaimStore, candidate_key # noqa: E402 @@ -57,6 +58,54 @@ class FakeClient: return None +class FakeDiscovery: + def __init__(self, count: int) -> None: + self.models = [ + HFModelSummary( + repo_id=f"owner/model-{index}", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + for index in range(count) + ] + + def list_recent_models(self, **_kwargs): # noqa: ANN003 + return self.models + + @staticmethod + def inspect_model(model: HFModelSummary) -> ModelInspection: + return ModelInspection(repo_id=model.repo_id, weight_files=["model.safetensors"]) + + +class DuplicateThenSuccessClient: + def __init__(self) -> None: + self.calls = 0 + self.active = 0 + + @staticmethod + def begin_cycle() -> None: + return None + + def available_submit_slots(self) -> int: + return max(0, 2 - self.active) + + @staticmethod + def list_tasks_page(**_kwargs) -> dict: # noqa: ANN003 + return {"code": 0, "data": {"records": [], "pages": 0}} + + @staticmethod + def processed_gpus_for_model(_model_id: str) -> set[str]: + return set() + + def add_task(self, _payload: dict) -> dict: + self.calls += 1 + if self.calls <= 2: + raise ModelHubAPIError("模型正在验证中，请勿重复提交") + self.active += 1 + return {"code": 0, "data": {"id": f"task-{self.calls}"}} + + def make_candidate(index: int) -> dict: return { "repoId": f"owner/model-{index}", @@ -66,6 +115,61 @@ def make_candidate(index: int) -> dict: class ClientPoolConcurrencyTests(unittest.TestCase): + def test_duplicate_batch_is_replaced_until_available_slots_are_filled(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + args = build_parser().parse_args( + [ + "--gpus", + "Iluvatar_bi-150", + "--task-types", + "text-generation", + "--limit", + "4", + "--max-scan-models", + "4", + "--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.history_archive_path = str(root / "history.jsonl") + client = DuplicateThenSuccessClient() + + summary = run_submission( + args, + now=datetime(2026, 1, 1, 12, tzinfo=timezone.utc), + hf_discovery=FakeDiscovery(4), # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + ) + + self.assertEqual(2, summary["targetSubmitCount"]) + self.assertEqual(4, summary["plannedSubmitCount"]) + self.assertEqual(2, summary["duplicateCount"]) + self.assertEqual(2, summary["submittedCount"]) + self.assertEqual(0, client.available_submit_slots()) + + def test_platform_duplicate_is_classified_as_skipped_candidate(self) -> None: + error = ModelHubAPIError("模型正在验证中，请勿重复提交") + self.assertTrue(is_duplicate_submission_error(error)) + + class DuplicateClient: + @staticmethod + def add_task(_payload): # noqa: ANN001 + raise error + + candidate = { + **make_candidate(1), + "taskType": "text-generation", + "framework": "vllm", + "configParams": "framework: vllm", + } + result = submit_candidate(candidate, DuplicateClient()) # type: ignore[arg-type] + self.assertEqual("duplicate", result["outcome"]) + def test_every_account_reaches_capacity_under_load(self) -> None: clients = [FakeClient() for _ in range(12)] pool = ModelHubClientPool( @@ -129,6 +233,21 @@ class ClientPoolConcurrencyTests(unittest.TestCase): class ProcessCoordinationTests(unittest.TestCase): + def test_duplicate_claims_are_retained_so_next_batch_moves_forward(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + path = Path(temporary_dir) / "claims.jsonl" + candidates = [make_candidate(index) for index in range(4)] + store = SubmissionClaimStore(path, owner_id="worker-a") + + duplicates = store.claim(candidates, limit=2) + store.mark_submitted(duplicates) + replacements = store.claim(candidates, limit=2) + + self.assertEqual( + {candidate_key(candidate) for candidate in candidates[2:]}, + {candidate_key(candidate) for candidate in replacements}, + ) + def test_concurrent_claim_stores_select_disjoint_candidates(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: path = Path(temporary_dir) / "claims.jsonl"