fix: refill from expanded candidate history

modelhub_submmit_api/main.py

@@ -2,6 +2,7 @@ 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
@@ -27,6 +28,8 @@ 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:
@@ -284,7 +287,7 @@ def process_model_for_candidates(
             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": "previously_failed_locally"})
+            skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": "local_failure_cooldown_24h"})
             continue
 
         compatible_task_types = [
@@ -321,6 +324,123 @@ def process_model_for_candidates(
     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,
+    read_concurrency: int,
+) -> 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,
+                ): 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,
@@ -509,6 +629,8 @@ def run_submission(
             "historyArchiveRecordCount": len(archived_history),
             "gpuStrategy": strategy_summary,
             "scanLimit": 0,
+            "candidateGoal": 0,
+            "scanStages": [],
             "scannedModels": 0,
             "candidateCount": 0,
             "targetSubmitCount": 0,
@@ -517,6 +639,7 @@ def run_submission(
             "submittedCount": 0,
             "duplicateCount": 0,
             "skippedCount": 0,
+            "skipReasonCounts": {},
             "failedCount": 0,
             "warnings": report.get("warnings", []),
             "runDir": str(run_dir),
@@ -563,55 +686,84 @@ def run_submission(
         platform_available_slots=platform_available_slots,
     )
 
-    model_query_kwargs = {
-        "pipeline_tags": pipeline_tags_for_task_types(selected_task_types),
-        "limit": scan_limit,
-        "min_downloads": args.min_downloads,
-        "updated_after": updated_after,
-    }
-    try:
-        model_query_kwargs["read_concurrency"] = max(1, args.read_concurrency)
-        models = hf_discovery.list_recent_models(**model_query_kwargs)
-    except TypeError:
-        models = hf_discovery.list_recent_models(
-            pipeline_tags=model_query_kwargs["pipeline_tags"],
-            limit=model_query_kwargs["limit"],
-            min_downloads=model_query_kwargs["min_downloads"],
-            updated_after=model_query_kwargs["updated_after"],
-        )
-
     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)
 
-    with ThreadPoolExecutor(max_workers=max(1, args.read_concurrency)) 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,
-            ): index
-            for index, model in enumerate(models)
+    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,
         }
-        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 = models[index]
-            try:
-                ordered_results[index] = future.result()
-            except Exception as exc:
-                ordered_results[index] = ([], [], [{"repoId": model.repo_id, "reason": str(exc)}])
+        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)
 
-        for index in range(len(models)):
-            model_candidates, model_skipped, model_failed = ordered_results.get(index, ([], [], []))
-            candidates.extend(model_candidates)
-            skipped.extend(model_skipped)
-            failed.extend(model_failed)
+        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=target_gpus,
+            selected_task_types=selected_task_types,
+            outcome_tracker=outcome_tracker,
+            read_concurrency=max(1, args.read_concurrency),
+        )
+        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]] = []
@@ -767,6 +919,8 @@ def run_submission(
         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)
 
     summary = {
         "generatedAt": now.isoformat(),
@@ -785,7 +939,9 @@ def run_submission(
         "historyArchiveRecordCount": len(archived_history),
         "gpuStrategy": strategy_summary,
         "scanLimit": scan_limit,
-        "scannedModels": len(models),
+        "candidateGoal": candidate_goal,
+        "scanStages": scan_stages,
+        "scannedModels": scanned_model_count,
         "candidateCount": len(candidates),
         "targetSubmitCount": target_submit_count,
         "maxSubmitAttempts": 0 if args.dry_run else min(
@@ -796,7 +952,9 @@ def run_submission(
         "duplicateCount": len(duplicate_candidates),
         "submittedCount": len(submitted),
         "skippedCount": len(skipped),
+        "skipReasonCounts": dict(skip_reason_counts.most_common()),
         "failedCount": len(failed),
+        "failureReasonCounts": dict(failure_reason_counts.most_common()),
         "submitConcurrencyUsed": submit_workers,
         "warnings": report.get("warnings", []),
         "runDir": str(run_dir),