submmit/modelhub_submmit_api/main.py

from __future__ import annotations

import argparse
import os
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import timedelta
from pathlib import Path
from typing import Any

from common import parse_datetime, runtime_instance_id, utc_now, write_json, write_jsonl
from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH, GPUStrategyManager
from hf_discovery import HuggingFaceDiscovery
from history_stats import (
    append_ledger_entry,
    build_empty_pre_submit_report,
    count_submissions_for_day,
    load_ledger,
    update_history_archive,
)
from market_intelligence import (
    DEFAULT_FETCH_WORKERS,
    DEFAULT_FRAMEWORK_MIN_SAMPLES,
    DEFAULT_FRAMEWORK_REFRESH_SECONDS,
    DEFAULT_MARKET_INTELLIGENCE_PATH,
    DEFAULT_QUEUE_REFRESH_SECONDS,
    DEFAULT_THROUGHPUT_WINDOW_HOURS,
    MarketIntelligenceManager,
)
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, compatible_frameworks_for_task, pipeline_tags_for_task_types, task_specs_for_model
from template_selector import TemplateSelector


DEFAULT_RUNS_DIR = Path("runs")
DEFAULT_LEDGER_PATH = Path("ledger/submissions.jsonl")
ADAPTIVE_SCAN_MAX_MODELS = 3000
ADAPTIVE_SCAN_MIN_FALLBACK_MODELS = 500


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Auto discover and submit public ModelScope models to ModelHub.")
    parser.add_argument("--gpu", help="Single GPU alias or platform name, for example: k100")
    parser.add_argument("--gpus", help="Comma-separated GPU aliases/platform names. Omit to auto-use all safe GPUs from templates.")
    parser.add_argument("--task-types", help="Comma-separated ModelHub task types. Omit to auto-enable all supported task types.")
    parser.add_argument("--limit", type=int, default=300, help="Maximum number of ModelScope models to scan per pipeline tag")
    parser.add_argument("--max-scan-models", type=int, default=0, help="Hard cap on total scanned models (0 means auto)")
    parser.add_argument(
        "--scan-multiplier",
        type=int,
        default=4,
        help="Multiplier used when auto-deriving scan limit from quota/queue capacity",
    )
    parser.add_argument("--min-downloads", type=int, default=50, help="Minimum downloads threshold")
    parser.add_argument("--daily-target", type=int, default=0, help="Daily submission target across all auto runs; 0 means unlimited")
    parser.add_argument("--dry-run", action="store_true", help="Only write artifacts without creating tasks")
    parser.add_argument(
        "--disable-gpu-strategy",
        action="store_true",
        help="Disable adaptive 70/30 proven-GPU scheduling and keep the legacy candidate order",
    )
    parser.add_argument(
        "--disable-market-intelligence",
        action="store_true",
        help="Disable live queue/throughput and public framework statistics",
    )
    parser.add_argument(
        "--gpu-strategy-refresh-submissions",
        type=int,
        default=200,
        help="Recalculate adaptive GPU choices after this many accepted submissions",
    )
    parser.add_argument("--skip-outcome-sync", action="store_true", help="Skip outcome sync from ModelHub before scanning")
    parser.add_argument("--skip-history-archive", action="store_true", help="Skip historical task archive download for this run")
    group = parser.add_mutually_exclusive_group()
    group.add_argument("--since-hours", type=int, help="Only scan models updated within the last N hours")
    group.add_argument("--updated-after", help="Only scan models updated after the given UTC time")
    parser.add_argument("--stats-window-days", type=int, default=7, help="History window in days")
    parser.add_argument(
        "--history-stats-threshold",
        type=int,
        default=500,
        help="Minimum local ledger records before using history stats to rank submissions",
    )
    parser.add_argument("--read-concurrency", type=int, default=4, help="Concurrency for read-only remote calls")
    parser.add_argument(
        "--submit-concurrency",
        type=int,
        default=0,
        help="Concurrency for task submission calls (0 = auto based on token/client count)",
    )
    parser.add_argument(
        "--max-submits-per-run",
        type=int,
        default=0,
        help="Maximum tasks to submit in one run (0 means unlimited)",
    )
    parser.add_argument("--runs-dir", default=str(DEFAULT_RUNS_DIR), help=argparse.SUPPRESS)
    parser.add_argument("--ledger-path", default=str(DEFAULT_LEDGER_PATH), help=argparse.SUPPRESS)
    parser.add_argument("--outcomes-path", default=str(DEFAULT_OUTCOMES_PATH), help=argparse.SUPPRESS)
    parser.add_argument(
        "--claims-path",
        default=os.getenv("MODELHUB_AGENT_CLAIMS_PATH", str(DEFAULT_CLAIMS_PATH)),
        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(
        "--submission-exclusions-path",
        default=os.getenv("MODELHUB_SUBMISSION_EXCLUSIONS_PATH", str(DEFAULT_SUBMISSION_EXCLUSIONS_PATH)),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--market-intelligence-state-path",
        default=os.getenv("MODELHUB_MARKET_INTELLIGENCE_PATH", str(DEFAULT_MARKET_INTELLIGENCE_PATH)),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--market-queue-refresh-seconds",
        type=int,
        default=int(os.getenv("MODELHUB_MARKET_QUEUE_REFRESH_SECONDS", str(DEFAULT_QUEUE_REFRESH_SECONDS))),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--market-framework-refresh-seconds",
        type=int,
        default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_REFRESH_SECONDS", str(DEFAULT_FRAMEWORK_REFRESH_SECONDS))),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--market-throughput-window-hours",
        type=int,
        default=int(os.getenv("MODELHUB_MARKET_THROUGHPUT_WINDOW_HOURS", str(DEFAULT_THROUGHPUT_WINDOW_HOURS))),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--market-fetch-workers",
        type=int,
        default=int(os.getenv("MODELHUB_MARKET_FETCH_WORKERS", str(DEFAULT_FETCH_WORKERS))),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--market-framework-min-samples",
        type=int,
        default=int(os.getenv("MODELHUB_MARKET_FRAMEWORK_MIN_SAMPLES", str(DEFAULT_FRAMEWORK_MIN_SAMPLES))),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--gpu-strategy-state-path",
        default=os.getenv("MODELHUB_GPU_STRATEGY_STATE_PATH", str(DEFAULT_GPU_STRATEGY_PATH)),
        help=argparse.SUPPRESS,
    )
    parser.add_argument("--gpu-strategy-recent-window", type=int, default=1000, help=argparse.SUPPRESS)
    parser.add_argument("--gpu-strategy-min-long-samples", type=int, default=100, help=argparse.SUPPRESS)
    parser.add_argument(
        "--capacity-state-path",
        default=os.getenv("MODELHUB_CAPACITY_STATE_PATH", str(DEFAULT_CAPACITY_STATE_PATH)),
        help=argparse.SUPPRESS,
    )
    parser.add_argument(
        "--capacity-probe-interval-cycles",
        type=int,
        default=int(os.getenv("MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES", "3")),
        help=argparse.SUPPRESS,
    )
    parser.add_argument("--hf-base-url", default=os.getenv("MODELSCOPE_BASE_URL", "https://modelscope.cn"), help=argparse.SUPPRESS)
    parser.add_argument("--modelhub-base-url", default=os.getenv("MODELHUB_BASE_URL", "https://modelhub.org.cn"), help=argparse.SUPPRESS)
    parser.add_argument("--modelhub-token", default=os.getenv("MODELHUB_XC_TOKEN") or os.getenv("XC_TOKEN"), help=argparse.SUPPRESS)
    return parser


def determine_updated_after(args: argparse.Namespace, now) -> Any:
    if args.since_hours is not None:
        return now - timedelta(hours=args.since_hours)
    if args.updated_after:
        parsed = parse_datetime(args.updated_after)
        if parsed is None:
            raise ValueError(f"Invalid --updated-after value: {args.updated_after}")
        return parsed
    return None


def resolve_task_types(args: argparse.Namespace) -> list[str]:
    if not args.task_types:
        return all_task_types()
    selected = [value.strip() for value in args.task_types.split(",") if value.strip()]
    unknown = [task_type for task_type in selected if task_type not in TASK_SPEC_BY_TYPE]
    if unknown:
        raise ValueError(f"Unknown task types: {', '.join(sorted(unknown))}")
    return selected


def resolve_target_gpus(args: argparse.Namespace, selector: TemplateSelector, task_types: list[str]) -> list[str]:
    raw_values: list[str] = []
    if args.gpus:
        raw_values.extend(value.strip() for value in args.gpus.split(",") if value.strip())
    if args.gpu:
        raw_values.append(args.gpu)

    if raw_values:
        normalized: list[str] = []
        for value in raw_values:
            target_gpu = selector.normalize_gpu(value)
            if target_gpu not in normalized:
                normalized.append(target_gpu)
        return normalized

    discovered: list[str] = []
    for task_type in task_types:
        for target_gpu in selector.supported_target_gpus(task_type, auto_only=True):
            if target_gpu not in discovered:
                discovered.append(target_gpu)
    return discovered


def resolve_scan_limit(
    args: argparse.Namespace,
    *,
    target_gpu_count: int,
    remaining_daily_quota: int,
    platform_available_slots: int | None,
) -> int:
    base_limit = max(1, args.limit)
    explicit_max = max(0, int(getattr(args, "max_scan_models", 0) or 0))
    if explicit_max > 0:
        return min(base_limit, explicit_max)

    multiplier = max(1, int(getattr(args, "scan_multiplier", 4) or 1))
    if args.daily_target > 0:
        needed = max(1, remaining_daily_quota)
    elif platform_available_slots is not None and platform_available_slots > 0:
        needed = max(1, platform_available_slots)
    else:
        needed = max(200, target_gpu_count * 20)

    max_submits_per_run = max(0, int(getattr(args, "max_submits_per_run", 0) or 0))
    if max_submits_per_run > 0:
        needed = min(needed, max_submits_per_run)

    return min(base_limit, needed * multiplier)


def choose_candidate_for_gpu(
    *,
    model: HFModelSummary,
    inspection: ModelInspection,
    template_selector: TemplateSelector,
    task_types: list[str],
    target_gpu: str,
    market_intelligence: MarketIntelligenceManager | None = None,
) -> CandidateModel | None:
    for task_type in task_types:
        supported_frameworks = template_selector.supported_frameworks_for_auto(task_type, target_gpu)
        compatible_frameworks: list[str] = []
        try:
            if supported_frameworks:
                compatible_frameworks = compatible_frameworks_for_task(
                    task_type,
                    target_gpu,
                    supported_frameworks,
                    inspection,
                )
        except ValueError:
            pass
        if market_intelligence is not None:
            compatible_frameworks = market_intelligence.selectable_frameworks(
                task_type=task_type,
                target_gpu=target_gpu,
                incumbent_frameworks=compatible_frameworks,
                inspection=inspection,
            )
        if not compatible_frameworks:
            continue

        for framework in compatible_frameworks:
            config_params = (
                market_intelligence.official_config(
                    task_type=task_type,
                    target_gpu=target_gpu,
                    framework=framework,
                    gguf_filename=inspection.selected_gguf,
                )
                if market_intelligence is not None
                else None
            )
            if config_params is not None:
                template_id = f"modelhub-live-{task_type}-{framework}-{target_gpu}".lower().replace("_", "-")
            else:
                try:
                    template = template_selector.select_template(task_type, framework, target_gpu)
                    config_params = template_selector.render_config(
                        template,
                        gguf_filename=inspection.selected_gguf if framework == "llamacpp" else None,
                    )
                    template_id = template.template_id
                except (KeyError, ValueError):
                    continue
            framework_metadata = (
                market_intelligence.framework_metadata(task_type, target_gpu, framework)
                if market_intelligence is not None
                else {}
            )
            score = float(framework_metadata.get("frameworkCombinedScore") or 0.0)
            warnings: list[str] = []
            if bool(framework_metadata.get("frameworkEvidenceQualified", False)):
                warnings.append("framework_selected_from_success_evidence")
            if config_params is not None and bool(framework_metadata.get("frameworkOfficialConfigValid", False)):
                warnings.append("official_build_config_synced")
            spec = TASK_SPEC_BY_TYPE[task_type]
            return CandidateModel(
                repo_id=model.repo_id,
                model_address=model.model_address,
                pipeline_tag=model.pipeline_tag,
                modality=spec.modality,
                task_type=task_type,
                target_gpu=target_gpu,
                framework=framework,
                template_id=template_id,
                config_params=config_params,
                downloads=model.downloads,
                last_modified=model.last_modified,
                gguf_filename=inspection.selected_gguf,
                score=score,
                warnings=warnings,
            )
    return None


def resolve_submit_concurrency(
    args: argparse.Namespace,
    *,
    modelhub_client: ModelHubClient | ModelHubClientPool | Any,
    planned_submit_count: int,
) -> int:
    explicit_workers = int(getattr(args, "submit_concurrency", 0) or 0)
    if explicit_workers > 0:
        return explicit_workers

    clients = getattr(modelhub_client, "clients", None)
    if isinstance(clients, list) and clients:
        if planned_submit_count > 0:
            return max(1, min(len(clients), planned_submit_count))
        return len(clients)

    return 1


def resolve_max_submit_count(
    args: argparse.Namespace,
    planned_count: int,
    remaining_daily_quota: int,
) -> int:
    explicit_limit = int(getattr(args, "max_submits_per_run", 0) or 0)
    planned_submit_count = min(planned_count, max(0, remaining_daily_quota))
    if explicit_limit > 0:
        planned_submit_count = min(planned_submit_count, explicit_limit)
    return planned_submit_count


def process_model_for_candidates(
    *,
    model: HFModelSummary,
    hf_discovery: HuggingFaceDiscovery,
    modelhub_client: ModelHubClient,
    template_selector: TemplateSelector,
    target_gpus: list[str],
    allowed_task_types: list[str],
    outcome_tracker: OutcomeTracker | None = None,
    submission_exclusion_store: SubmissionExclusionStore | None = None,
    market_intelligence: MarketIntelligenceManager | 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'}"}], []

    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]] = []
    allowed_task_types_set = [spec.task_type for spec in specs]
    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
        if outcome_tracker and outcome_tracker.is_model_gpu_failed(model.repo_id, target_gpu):
            skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"})
            continue

        compatible_task_types = [
            task_type
            for task_type in allowed_task_types_set
            if template_selector.supported_frameworks_for_auto(task_type, target_gpu)
        ]
        if not compatible_task_types:
            skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "no_compatible_auto_template_or_framework"})
            continue
        pending_task_types_by_gpu.append((target_gpu, compatible_task_types))

    if not pending_task_types_by_gpu:
        return candidates, skipped, failed

    try:
        inspection = hf_discovery.inspect_model(model)
    except Exception as exc:
        return [], skipped, [{"repoId": model.repo_id, "reason": str(exc)}]

    for target_gpu, task_types in pending_task_types_by_gpu:
        best = choose_candidate_for_gpu(
            model=model,
            inspection=inspection,
            template_selector=template_selector,
            task_types=task_types,
            target_gpu=target_gpu,
            market_intelligence=market_intelligence,
        )
        if best is None:
            reason = (
                "no_publicly_vetted_compatible_framework"
                if market_intelligence is not None
                else "no_compatible_auto_template_or_framework"
            )
            skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason})
            continue
        record = candidate_to_record(best)
        if market_intelligence is not None:
            record.update(market_intelligence.gpu_metadata(target_gpu))
            record.update(market_intelligence.framework_metadata(best.task_type, target_gpu, best.framework))
        candidates.append(record)

    return candidates, skipped, failed


def build_adaptive_scan_stages(
    *,
    now,
    initial_updated_after,
    initial_limit: int,
    max_models: int = ADAPTIVE_SCAN_MAX_MODELS,
) -> list[dict[str, Any]]:
    max_models = max(1, min(ADAPTIVE_SCAN_MAX_MODELS, int(max_models)))
    initial_limit = min(max_models, max(1, int(initial_limit)))
    stages: list[dict[str, Any]] = [
        {
            "name": "configured_window",
            "updatedAfter": initial_updated_after,
            "limit": initial_limit,
        }
    ]

    seven_days_ago = now - timedelta(days=7)
    thirty_days_ago = now - timedelta(days=30)
    if initial_updated_after is not None and initial_updated_after > seven_days_ago:
        stages.append(
            {
                "name": "last_7_days",
                "updatedAfter": seven_days_ago,
                "limit": min(max_models, max(initial_limit, ADAPTIVE_SCAN_MIN_FALLBACK_MODELS)),
            }
        )
    if initial_updated_after is not None and initial_updated_after > thirty_days_ago:
        stages.append(
            {
                "name": "last_30_days",
                "updatedAfter": thirty_days_ago,
                "limit": min(max_models, max(initial_limit, 1500)),
            }
        )
    if initial_updated_after is not None or initial_limit < max_models:
        stages.append(
            {
                "name": "all_history",
                "updatedAfter": None,
                "limit": max_models,
            }
        )

    deduped: list[dict[str, Any]] = []
    seen: set[tuple[str | None, int]] = set()
    for stage in stages:
        updated_after = stage["updatedAfter"]
        signature = (updated_after.isoformat() if updated_after is not None else None, int(stage["limit"]))
        if signature in seen:
            continue
        seen.add(signature)
        deduped.append(stage)
    return deduped


def collect_candidates_from_models(
    *,
    models: list[HFModelSummary],
    seen_model_ids: set[str],
    candidate_goal: int,
    hf_discovery: HuggingFaceDiscovery,
    modelhub_client: ModelHubClient,
    template_selector: TemplateSelector,
    target_gpus: list[str],
    selected_task_types: list[str],
    outcome_tracker: OutcomeTracker | None,
    submission_exclusion_store: SubmissionExclusionStore | None,
    read_concurrency: int,
    market_intelligence: MarketIntelligenceManager | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], int]:
    candidates: list[dict[str, Any]] = []
    skipped: list[dict[str, Any]] = []
    failed: list[dict[str, Any]] = []
    processed_count = 0
    workers = max(1, int(read_concurrency))
    chunk_size = max(16, workers * 8)
    new_models = [model for model in models if model.repo_id not in seen_model_ids]

    for offset in range(0, len(new_models), chunk_size):
        if len(candidates) >= candidate_goal:
            break
        chunk = new_models[offset : offset + chunk_size]
        for model in chunk:
            seen_model_ids.add(model.repo_id)

        with ThreadPoolExecutor(max_workers=workers) as executor:
            futures = {
                executor.submit(
                    process_model_for_candidates,
                    model=model,
                    hf_discovery=hf_discovery,
                    modelhub_client=modelhub_client,
                    template_selector=template_selector,
                    target_gpus=target_gpus,
                    allowed_task_types=selected_task_types,
                    outcome_tracker=outcome_tracker,
                    submission_exclusion_store=submission_exclusion_store,
                    market_intelligence=market_intelligence,
                ): index
                for index, model in enumerate(chunk)
            }
            ordered_results: dict[int, tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]] = {}
            for future in as_completed(futures):
                index = futures[future]
                model = chunk[index]
                try:
                    ordered_results[index] = future.result()
                except Exception as exc:
                    ordered_results[index] = ([], [], [{"repoId": model.repo_id, "reason": str(exc)}])

        for index in range(len(chunk)):
            model_candidates, model_skipped, model_failed = ordered_results.get(index, ([], [], []))
            candidates.extend(model_candidates)
            skipped.extend(model_skipped)
            failed.extend(model_failed)
        processed_count += len(chunk)

    return candidates, skipped, failed, processed_count


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"],
        "targetGpu": candidate["targetGpu"],
        "framework": candidate["framework"],
        "configParams": candidate["configParams"],
    }
    strategy_id = os.getenv("STRATEGY_ID")
    if strategy_id:
        payload["strategyId"] = strategy_id
    submit_time = utc_now()
    try:
        response = modelhub_client.add_task(payload)
        task_id = extract_task_id_from_submit_response(response)
        if task_id is None:
            task_id = modelhub_client.find_recent_task_id(candidate["repoId"], candidate["targetGpu"], submit_time)
        return {
            "outcome": "submitted",
            "candidate": candidate,
            "submitTime": submit_time.isoformat(),
            "taskId": task_id,
            "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']} "
                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}",
            flush=True,
        )
        return {
            "outcome": "failed",
            "candidate": candidate,
            "reason": str(exc),
        }


def make_run_dir(runs_dir: Path, now) -> Path:
    runs_dir.mkdir(parents=True, exist_ok=True)
    base_name = now.strftime("%Y%m%dT%H%M%SZ")
    for suffix in range(0, 1000):
        candidate = runs_dir / (base_name if suffix == 0 else f"{base_name}.{suffix:03d}")
        try:
            candidate.mkdir(exist_ok=False)
            return candidate
        except FileExistsError:
            continue
    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,
    *,
    now=None,
    hf_discovery: HuggingFaceDiscovery | None = None,
    modelhub_client: ModelHubClient | ModelHubClientPool | None = None,
    template_selector: TemplateSelector | None = None,
    outcome_tracker: OutcomeTracker | None = None,
) -> dict[str, Any]:
    now = now or utc_now()
    template_selector = template_selector or TemplateSelector()
    selected_task_types = resolve_task_types(args)
    target_gpus = resolve_target_gpus(args, template_selector, selected_task_types)
    if not target_gpus:
        raise RuntimeError("No auto-submittable GPUs are available for the selected task types")
    submission_target_gpus = list(target_gpus)

    hf_discovery = hf_discovery or HuggingFaceDiscovery(base_url=args.hf_base_url)
    if modelhub_client is None:
        modelhub_tokens = list(getattr(args, "modelhub_tokens", None) or ([] if not args.modelhub_token else [args.modelhub_token]))
        token_values: list[str | None] = modelhub_tokens or [args.modelhub_token]
        clients = [ModelHubClient(token=token, base_url=args.modelhub_base_url) for token in token_values]
        modelhub_client = ModelHubClientPool(
            clients,
            capacity_probe_interval_cycles=max(0, int(getattr(args, "capacity_probe_interval_cycles", 3) or 0)),
            capacity_state_path=Path(getattr(args, "capacity_state_path", DEFAULT_CAPACITY_STATE_PATH)),
        )

    if hasattr(modelhub_client, "begin_cycle"):
        modelhub_client.begin_cycle()

    runs_dir = Path(args.runs_dir)
    ledger_path = Path(args.ledger_path)
    history_archive_path = Path(args.history_archive_path)
    run_dir = make_run_dir(runs_dir, now)
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    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}
    market_intelligence: MarketIntelligenceManager | None = None
    market_summary: dict[str, Any] = {"enabled": False}
    strategy_enabled = (
        not bool(getattr(args, "disable_gpu_strategy", False))
        and not bool(getattr(args, "gpu", None) or getattr(args, "gpus", None))
        and len(target_gpus) > 1
    )
    synced_count = 0
    if not getattr(args, "skip_outcome_sync", False):
        try:
            synced_count = outcome_tracker.sync_from_api(modelhub_client)
        except Exception:
            pass

    updated_after = determine_updated_after(args, now)
    history_begin = now - timedelta(days=args.stats_window_days)
    ledger_entries = load_ledger(ledger_path)
    day_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    # Count today's submissions from the local ledger (avoids expensive paginated API call)
    daily_snapshot = count_submissions_for_day(tasks=[], ledger_entries=ledger_entries, day_start=day_start, day_end=now)
    # A configured daily target must include every pooled account and every
    # concurrent worker, not only the first token's first page.
    if args.daily_target > 0:
        list_kwargs: dict[str, Any] = {
            "page_size": 100,
            "only_mine": True,
            "begin_time": day_start,
            "end_time": now,
        }
        if isinstance(modelhub_client, ModelHubClientPool):
            list_kwargs["_fanout_all"] = True
        today_tasks = modelhub_client.list_tasks(**list_kwargs)
        daily_snapshot = count_submissions_for_day(
            tasks=today_tasks,
            ledger_entries=ledger_entries,
            day_start=day_start,
            day_end=now,
        )
    # In unlimited mode, keep the inexpensive one-page fallback for an empty ledger.
    elif daily_snapshot["totalCount"] <= 0:
        today_tasks_page = modelhub_client.list_tasks_page(
            current=1,
            page_size=20,
            only_mine=True,
            begin_time=day_start,
            end_time=now,
        )
        today_tasks = (today_tasks_page.get("data") or {}).get("records") or []
        daily_snapshot = count_submissions_for_day(tasks=today_tasks, ledger_entries=ledger_entries, day_start=day_start, day_end=now)
    platform_available_slots = None
    if hasattr(modelhub_client, "available_submit_slots"):
        platform_available_slots = max(0, int(modelhub_client.available_submit_slots()))
    unlimited_daily_target = args.daily_target <= 0
    if unlimited_daily_target:
        remaining_daily_quota = platform_available_slots if platform_available_slots is not None else 1_000_000_000
    else:
        remaining_daily_quota = max(0, args.daily_target - daily_snapshot["totalCount"])
        if platform_available_slots is not None:
            remaining_daily_quota = min(remaining_daily_quota, platform_available_slots)

    history_report_reason = "history_archive_skipped" if getattr(args, "skip_history_archive", False) else "history_archive_only_mode"
    if args.daily_target > 0 and remaining_daily_quota <= 0:
        archived_history: list[dict[str, Any]] = []
        report = build_empty_pre_submit_report(
            window_days=args.stats_window_days,
            generated_at=now,
            reason="daily_target_reached_before_scan",
            ledger_entries=len(ledger_entries),
        )
        write_json(run_dir / "pre_submit_report.json", report)
        return {
            "generatedAt": now.isoformat(),
            "dryRun": bool(args.dry_run),
            "selectedTaskTypes": selected_task_types,
            "targetGpus": target_gpus,
            "submissionEligibleGpus": submission_target_gpus,
            "dailyTarget": args.daily_target,
            "unlimitedDailyTarget": unlimited_daily_target,
            "submittedTodayBeforeRun": daily_snapshot["totalCount"],
            "remainingDailyQuotaBeforeRun": remaining_daily_quota,
            "platformAvailableSlotsBeforeRun": platform_available_slots,
            "historyStatsEnabled": False,
            "historyStatsThreshold": getattr(args, "history_stats_threshold", 500),
            "historyArchivePath": str(history_archive_path),
            "historyArchiveRecordCount": len(archived_history),
            "gpuStrategy": strategy_summary,
            "marketIntelligence": market_summary,
            "scanLimit": 0,
            "candidateGoal": 0,
            "scanStages": [],
            "scannedModels": 0,
            "candidateCount": 0,
            "candidateFrameworkCounts": {},
            "targetSubmitCount": 0,
            "maxSubmitAttempts": 0,
            "plannedSubmitCount": 0,
            "submittedCount": 0,
            "submittedGpuCounts": {},
            "submittedFrameworkCounts": {},
            "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,
        }

    market_capable = (
        hasattr(modelhub_client, "list_machine_info")
        and hasattr(modelhub_client, "list_framework_stats")
        and hasattr(modelhub_client, "list_tasks_page")
    )
    if not bool(getattr(args, "disable_market_intelligence", False)) and market_capable:
        market_intelligence = MarketIntelligenceManager(
            Path(getattr(args, "market_intelligence_state_path", DEFAULT_MARKET_INTELLIGENCE_PATH)),
            queue_refresh_seconds=max(
                60,
                int(getattr(args, "market_queue_refresh_seconds", DEFAULT_QUEUE_REFRESH_SECONDS) or DEFAULT_QUEUE_REFRESH_SECONDS),
            ),
            framework_refresh_seconds=max(
                300,
                int(
                    getattr(args, "market_framework_refresh_seconds", DEFAULT_FRAMEWORK_REFRESH_SECONDS)
                    or DEFAULT_FRAMEWORK_REFRESH_SECONDS
                ),
            ),
            throughput_window_hours=max(
                1,
                int(
                    getattr(args, "market_throughput_window_hours", DEFAULT_THROUGHPUT_WINDOW_HOURS)
                    or DEFAULT_THROUGHPUT_WINDOW_HOURS
                ),
            ),
            fetch_workers=max(
                1,
                int(getattr(args, "market_fetch_workers", DEFAULT_FETCH_WORKERS) or DEFAULT_FETCH_WORKERS),
            ),
            framework_min_samples=max(
                1,
                int(
                    getattr(args, "market_framework_min_samples", DEFAULT_FRAMEWORK_MIN_SAMPLES)
                    or DEFAULT_FRAMEWORK_MIN_SAMPLES
                ),
            ),
        )
        market_intelligence.prepare(
            modelhub_client,
            supported_gpus=target_gpus,
            task_types=selected_task_types,
            now=now,
        )
        try:
            market_intelligence.set_local_outcome_stats(outcome_tracker.get_stats_report())
        except Exception:
            market_intelligence.set_local_outcome_stats(None)
        market_summary = market_intelligence.summary()
        submission_target_gpus = market_intelligence.eligible_gpus(target_gpus)
        market_summary["eligibleGpus"] = submission_target_gpus
        market_summary["shadowOnlyGpus"] = [gpu for gpu in target_gpus if gpu not in submission_target_gpus]
        print(
            f"[market] eligible_gpus={','.join(submission_target_gpus) or 'none'} "
            f"shadow_only={','.join(market_summary['shadowOnlyGpus']) or 'none'}",
            flush=True,
        )

    if strategy_enabled:
        strategy_manager = GPUStrategyManager(
            Path(getattr(args, "gpu_strategy_state_path", DEFAULT_GPU_STRATEGY_PATH)),
            refresh_submissions=max(1, int(getattr(args, "gpu_strategy_refresh_submissions", 200) or 200)),
            recent_terminal_window=max(1, int(getattr(args, "gpu_strategy_recent_window", 1000) or 1000)),
            long_term_min_samples=max(1, int(getattr(args, "gpu_strategy_min_long_samples", 100) or 100)),
            market_intelligence=market_intelligence,
        )
        strategy_manager.prepare(modelhub_client, supported_gpus=target_gpus, now=now)
        strategy_summary = strategy_manager.summary()

    if getattr(args, "skip_history_archive", False):
        archived_history = []
    else:
        history_tasks = modelhub_client.list_tasks(
            page_size=50,
            only_mine=True,
            begin_time=history_begin,
            end_time=now,
        )
        archived_history = update_history_archive(
            history_archive_path,
            history_tasks,
            limit=getattr(args, "history_archive_limit", 5000),
        )

    report = build_empty_pre_submit_report(
        window_days=args.stats_window_days,
        generated_at=now,
        reason=history_report_reason,
        ledger_entries=len(ledger_entries),
    )
    write_json(run_dir / "pre_submit_report.json", report)

    scan_limit = resolve_scan_limit(
        args,
        target_gpu_count=len(submission_target_gpus),
        remaining_daily_quota=remaining_daily_quota,
        platform_available_slots=platform_available_slots,
    )

    candidates: list[dict[str, Any]] = []
    skipped: list[dict[str, Any]] = []
    failed: list[dict[str, Any]] = []
    scan_stages: list[dict[str, Any]] = []
    seen_model_ids: set[str] = set()
    scanned_model_count = 0
    explicit_submit_limit = max(0, int(getattr(args, "max_submits_per_run", 0) or 0))
    if platform_available_slots is None and unlimited_daily_target:
        submission_goal = scan_limit
    else:
        submission_goal = max(0, int(remaining_daily_quota))
    if explicit_submit_limit > 0:
        submission_goal = min(submission_goal, explicit_submit_limit)
    if strategy_manager is not None:
        submission_goal = min(submission_goal, strategy_manager.submissions_until_refresh)
    attempt_multiplier = max(1, int(getattr(args, "scan_multiplier", 4) or 1))
    candidate_goal = max(submission_goal, submission_goal * attempt_multiplier)

    pipeline_tags = pipeline_tags_for_task_types(selected_task_types)
    explicit_scan_cap = max(0, int(getattr(args, "max_scan_models", 0) or 0))
    for stage in build_adaptive_scan_stages(
        now=now,
        initial_updated_after=updated_after,
        initial_limit=scan_limit,
        max_models=explicit_scan_cap or ADAPTIVE_SCAN_MAX_MODELS,
    ):
        if candidate_goal <= 0 or len(candidates) >= candidate_goal:
            break
        stage_updated_after = stage["updatedAfter"]
        stage_limit = int(stage["limit"])
        query_kwargs = {
            "pipeline_tags": pipeline_tags,
            "limit": stage_limit,
            "min_downloads": args.min_downloads,
            "updated_after": stage_updated_after,
        }
        try:
            models = hf_discovery.list_recent_models(
                **query_kwargs,
                read_concurrency=max(1, args.read_concurrency),
            )
        except TypeError:
            models = hf_discovery.list_recent_models(**query_kwargs)

        stage_candidates, stage_skipped, stage_failed, processed_count = collect_candidates_from_models(
            models=models,
            seen_model_ids=seen_model_ids,
            candidate_goal=max(1, candidate_goal - len(candidates)),
            hf_discovery=hf_discovery,
            modelhub_client=modelhub_client,
            template_selector=template_selector,
            target_gpus=submission_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),
            market_intelligence=market_intelligence,
        )
        candidates.extend(stage_candidates)
        skipped.extend(stage_skipped)
        failed.extend(stage_failed)
        scanned_model_count += processed_count
        stage_summary = {
            "name": stage["name"],
            "updatedAfter": stage_updated_after.isoformat() if stage_updated_after is not None else None,
            "requestedLimit": stage_limit,
            "discoveredModels": len(models),
            "newModelsProcessed": processed_count,
            "candidatesAdded": len(stage_candidates),
            "candidateCountAfterStage": len(candidates),
            "skippedAdded": len(stage_skipped),
            "failedAdded": len(stage_failed),
        }
        scan_stages.append(stage_summary)
        print(
            f"[scan] stage={stage['name']} discovered={len(models)} new_processed={processed_count} "
            f"candidates_added={len(stage_candidates)} candidates_total={len(candidates)}/{candidate_goal} "
            f"skipped_added={len(stage_skipped)}",
            flush=True,
        )

    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),
        remaining_daily_quota=remaining_daily_quota,
    )
    if strategy_manager is not None:
        target_submit_count = min(target_submit_count, strategy_manager.submissions_until_refresh)
    instance_id = runtime_instance_id()
    diversified_candidates = diversify_candidates(candidates, instance_id=instance_id)
    if strategy_manager is not None:
        diversified_candidates = strategy_manager.order_candidates(diversified_candidates)
    write_jsonl(run_dir / "candidates.jsonl", diversified_candidates)
    claim_store: SubmissionClaimStore | None = None
    attempted_candidates: list[dict[str, Any]] = []
    submit_workers = 1

    if args.dry_run:
        attempted_candidates = diversified_candidates[:target_submit_count]
    else:
        claim_store = SubmissionClaimStore(
            Path(getattr(args, "claims_path", DEFAULT_CLAIMS_PATH)),
            owner_id=instance_id,
        )
        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),
        )

        # 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
                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),
            )
            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_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)
                    skipped.append(
                        {
                            "repoId": candidate["repoId"],
                            "targetGpu": candidate["targetGpu"],
                            "reason": "already_validating_on_platform",
                        }
                    )
                    continue
                if result["outcome"] in {"failed", "precheck_deferred"}:
                    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,
                    {
                        "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, *batch_uniqueness_rejected_candidates]
            )
            claim_store.release(batch_failed_candidates)
            if strategy_manager is not None:
                strategy_manager.record_accepted(batch_submitted_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)
    write_jsonl(run_dir / "failed.jsonl", failed)

    if strategy_manager is not None:
        strategy_summary = strategy_manager.summary()

    outcome_tracker.save()
    skip_reason_counts = Counter(str(item.get("reason") or "unknown") for item in skipped)
    failure_reason_counts = Counter(str(item.get("reason") or "unknown") for item in failed)
    candidate_framework_counts = Counter(str(item.get("framework") or "unknown") for item in candidates)
    submitted_gpu_counts = Counter(str(item.get("targetGpu") or "unknown") for item in submitted)
    submitted_framework_counts = Counter(str(item.get("framework") or "unknown") for item in submitted)

    summary = {
        "generatedAt": now.isoformat(),
        "dryRun": bool(args.dry_run),
        "selectedTaskTypes": selected_task_types,
        "targetGpus": target_gpus,
        "submissionEligibleGpus": submission_target_gpus,
        "maxSubmitsPerRun": int(getattr(args, "max_submits_per_run", 0) or 0),
        "dailyTarget": args.daily_target,
        "unlimitedDailyTarget": unlimited_daily_target,
        "submittedTodayBeforeRun": daily_snapshot["totalCount"],
        "remainingDailyQuotaBeforeRun": remaining_daily_quota,
        "platformAvailableSlotsBeforeRun": platform_available_slots,
        "historyStatsEnabled": False,
        "historyStatsThreshold": getattr(args, "history_stats_threshold", 500),
        "historyArchivePath": str(history_archive_path),
        "historyArchiveRecordCount": len(archived_history),
        "gpuStrategy": strategy_summary,
        "marketIntelligence": market_summary,
        "scanLimit": scan_limit,
        "candidateGoal": candidate_goal,
        "scanStages": scan_stages,
        "scannedModels": scanned_model_count,
        "candidateCount": len(candidates),
        "candidateFrameworkCounts": dict(candidate_framework_counts.most_common()),
        "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),
        "modelGpuUniquenessRejectedCount": len(uniqueness_rejected_candidates),
        "submittedCount": len(submitted),
        "submittedGpuCounts": dict(submitted_gpu_counts.most_common()),
        "submittedFrameworkCounts": dict(submitted_framework_counts.most_common()),
        "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),
        "outcomeSyncCount": synced_count,
    }
    write_json(run_dir / "summary.json", summary)
    return summary


def candidate_to_record(candidate: CandidateModel) -> dict[str, Any]:
    return {
        "repoId": candidate.repo_id,
        "modelAddress": candidate.model_address,
        "pipelineTag": candidate.pipeline_tag,
        "modality": candidate.modality,
        "taskType": candidate.task_type,
        "taskPriority": TASK_SPEC_BY_TYPE[candidate.task_type].priority,
        "targetGpu": candidate.target_gpu,
        "framework": candidate.framework,
        "templateId": candidate.template_id,
        "configParams": candidate.config_params,
        "downloads": candidate.downloads,
        "lastModified": candidate.last_modified.isoformat() if candidate.last_modified else None,
        "ggufFilename": candidate.gguf_filename,
        "score": candidate.score,
        "warnings": candidate.warnings,
    }


def extract_task_id_from_submit_response(response: dict[str, Any]) -> str | None:
    data = response.get("data")
    if isinstance(data, dict) and data.get("id") is not None:
        return str(data["id"])
    return None


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    summary = run_submission(args)
    print(f"run_dir={summary['runDir']}")
    print(f"daily_target={summary['dailyTarget']} remaining_before_run={summary['remainingDailyQuotaBeforeRun']}")
    print(f"task_types={','.join(summary['selectedTaskTypes'])}")
    print(f"gpus={','.join(summary['targetGpus'])}")
    print(
        f"candidates={summary['candidateCount']} planned={summary['plannedSubmitCount']} "
        f"submitted={summary['submittedCount']} skipped={summary['skippedCount']} failed={summary['failedCount']}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())