diff --git a/README.md b/README.md index cdab283b..9a0d5e80 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,8 @@ Optional tuning: - `MODELHUB_AGENT_POLL_INTERVAL_SECONDS` default `15` - `MODELHUB_AGENT_IDLE_INTERVAL_SECONDS` default `60` - `MODELHUB_AGENT_POST_CYCLE_COOLDOWN_SECONDS` default `2` -- `MODELHUB_AGENT_MAX_SUBMITS_PER_RUN` default `0` (fill all currently available slots) +- The hosted entrypoint always uses `--max-submits-per-run 0` so stale deployment + settings cannot restrict a refill cycle to five submissions. - `MODELHUB_AGENT_ACTIVE_TASK_CAP` default `100` per account - `MODELHUB_CAPACITY_PROBE_INTERVAL_CYCLES` default `3` - `MODELHUB_CAPACITY_STATE_PATH` default `.modelhub_state/account_capacity.json` @@ -44,6 +45,7 @@ Optional tuning: - `MODELHUB_GPU_STRATEGY_STATE_PATH` default `.modelhub_state/gpu_strategy.json` - `MODELSCOPE_PAGE_INTERVAL_SECONDS` default `0.25` - `MODELSCOPE_PAGE_CACHE_TTL_SECONDS` default `900` +- `MODELHUB_AGENT_VERIFY_CACHE_TTL_SECONDS` default `900` ## Adaptive GPU Strategy @@ -63,6 +65,18 @@ ModelScope HTTP 429 responses use exponential backoff and `Retry-After`. Success pages remain cached, so a later cycle retries the failed page instead of restarting the whole pagination scan. +## Adaptive Candidate Discovery + +The configured recent window remains the fast path. If it contains no usable +model/GPU combinations, the same run progressively expands discovery to the last +7 days, the last 30 days, and finally older history (up to 3,000 models). Scanning +stops as soon as enough replacement candidates have been found. + +Model verification results are reused for 15 minutes across poll cycles, and a +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. + ## Concurrent Agents The token pool keeps a local reservation for every in-flight submission, so a @@ -80,7 +94,8 @@ 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.3` or newer includes -this behavior. +duplicate replacement behavior, while version `2026.08.02.4` adds adaptive +candidate-window expansion and skip-reason reporting. ## Deploy diff --git a/main.py b/main.py index c4103ee8..e7260a57 100644 --- a/main.py +++ b/main.py @@ -52,8 +52,6 @@ def _worker_command() -> list[str]: os.getenv("MODELHUB_AGENT_IDLE_INTERVAL_SECONDS", "60"), "--post-cycle-cooldown-seconds", os.getenv("MODELHUB_AGENT_POST_CYCLE_COOLDOWN_SECONDS", "2"), - "--max-submits-per-run", - os.getenv("MODELHUB_AGENT_MAX_SUBMITS_PER_RUN", "0"), "--skip-history-archive", ] @@ -70,6 +68,9 @@ def _worker_command() -> list[str]: cmd.extend(["--gpus", gpus]) cmd.extend(_csv_args("MODELHUB_AGENT_EXTRA_ARGS")) + # The hosted agent's goal is to fill every currently available slot. Keep + # this last so an old environment or extra-args value cannot restore "5". + cmd.extend(["--max-submits-per-run", "0"]) return cmd diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 37126c16..4ff4ccad 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -79,6 +79,10 @@ bash run_poll.sh --dry-run duplicates do not advance it. The next cycle refreshes platform history before submitting again. - Strategy state is stored in `.modelhub_state/gpu_strategy.json`; a generation never recalculates during candidate submission. +- Candidate discovery starts with the configured recent window, then automatically expands to + 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. - 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. @@ -88,6 +92,8 @@ bash run_poll.sh --dry-run on the next cycle. - Every third poll cycle, a full account gets one controlled capacity probe. A successful probe raises that account's persisted known limit; a capacity rejection enters cooldown. +- Each `[scan]` log records the discovery stage and candidate yield. The final `[daily] wave_done` + log includes `skip_reasons`, making empty candidate pools distinguishable from API failures. ## Important Flags diff --git a/modelhub_submmit_api/daily_runner.py b/modelhub_submmit_api/daily_runner.py index 0f09564a..dba8ee8e 100644 --- a/modelhub_submmit_api/daily_runner.py +++ b/modelhub_submmit_api/daily_runner.py @@ -236,11 +236,17 @@ def run_daily_batches( wave_results.append(wave_result) submitted_total += summary["submittedCount"] round_submitted += summary["submittedCount"] + skip_reasons = summary.get("skipReasonCounts") or {} + skip_reason_text = ",".join( + f"{reason}:{count}" + for reason, count in list(skip_reasons.items())[:4] + ) or "none" log( 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"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 e92663a3..7bfd70f6 100644 --- a/modelhub_submmit_api/main.py +++ b/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), diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index ddbab7ed..711de90d 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -321,7 +321,7 @@ class ModelHubClientPool: identity_hash = hashlib.sha256(identity.encode("utf-8")).hexdigest() self._selection_cursor = int(identity_hash[:12], 16) % len(clients) self._verify_cache: dict[str, tuple[float, dict[str, Any]]] = {} - self._verify_cache_ttl = max(1.0, float(os.getenv("MODELHUB_AGENT_VERIFY_CACHE_TTL_SECONDS", "30"))) + self._verify_cache_ttl = max(1.0, float(os.getenv("MODELHUB_AGENT_VERIFY_CACHE_TTL_SECONDS", "900"))) # Single reader client to avoid fanout on read operations self._reader = clients[0] @@ -539,12 +539,15 @@ class ModelHubClientPool: self._verify_cache[model_id] = (time.monotonic(), payload) def begin_cycle(self) -> None: - """Drop model verification cache entries from the previous scan cycle.""" + """Keep recent verification results across cycles and prune expired entries.""" with self._state_lock: - self._verify_cache.clear() + now = time.monotonic() + for model_id, (cached_at, _payload) in list(self._verify_cache.items()): + 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]: - # Check cache first (per-cycle cache to avoid repeated API calls for the same model) + # 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: diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index 79e36893..d0bd6dbc 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -23,7 +23,7 @@ class OutcomeTracker: self._records: list[dict[str, Any]] = [] self._by_task_id: dict[str, dict[str, Any]] = {} self._by_model_gpu: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) - self._failed_model_gpus: set[tuple[str, str]] = set() + self._failed_model_gpus: dict[tuple[str, str], datetime] = {} self._records = read_jsonl(self.path) self._rebuild_indexes() @@ -117,8 +117,19 @@ class OutcomeTracker: return updated_count - def is_model_gpu_failed(self, model_id: str, target_gpu: str) -> bool: - return (model_id, target_gpu) in self._failed_model_gpus + def is_model_gpu_failed( + self, + model_id: str, + target_gpu: str, + *, + cooldown_hours: int = 24, + now: datetime | None = None, + ) -> bool: + failed_at = self._failed_model_gpus.get((model_id, target_gpu)) + if failed_at is None: + return False + now = now or utc_now() + return failed_at >= now - timedelta(hours=max(0, int(cooldown_hours))) def get_stats_report(self) -> dict[str, Any]: now = _now_iso() @@ -186,12 +197,21 @@ class OutcomeTracker: def _rebuild_failed_index(self) -> None: self._failed_model_gpus.clear() + latest_by_combo: dict[tuple[str, str], tuple[datetime, dict[str, Any]]] = {} for record in self._records: + model_id = record.get("modelId") or "" + target_gpu = record.get("targetGpu") or "" + event_time = parse_datetime(record.get("lastSyncTime")) or parse_datetime(record.get("submitTime")) + if not model_id or not target_gpu or event_time is None: + continue + key = (model_id, target_gpu) + current = latest_by_combo.get(key) + if current is None or event_time >= current[0]: + latest_by_combo[key] = (event_time, record) + + for key, (event_time, record) in latest_by_combo.items(): if record.get("outcome") == "failed": - model_id = record.get("modelId") or "" - target_gpu = record.get("targetGpu") or "" - if model_id and target_gpu: - self._failed_model_gpus.add((model_id, target_gpu)) + self._failed_model_gpus[key] = event_time @staticmethod def _update_record_from_task(record: dict[str, Any], task: dict[str, Any]) -> None: diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 0a66a1f2..d166abb3 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.02.3" +AGENT_VERSION = "2026.08.02.4" diff --git a/tests/test_agent_entrypoint.py b/tests/test_agent_entrypoint.py new file mode 100644 index 00000000..636c34b1 --- /dev/null +++ b/tests/test_agent_entrypoint.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import importlib.util +import os +import unittest +from pathlib import Path +from unittest.mock import patch + + +ROOT_DIR = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location("modelhub_agent_entrypoint", ROOT_DIR / "main.py") +if SPEC is None or SPEC.loader is None: + raise RuntimeError("Unable to load the hosted agent entrypoint") +ENTRYPOINT = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(ENTRYPOINT) + + +class HostedAgentEntrypointTests(unittest.TestCase): + def test_hosted_worker_always_uses_unlimited_cycle_submissions(self) -> None: + with patch.dict( + os.environ, + { + "MODELHUB_AGENT_MAX_SUBMITS_PER_RUN": "5", + "MODELHUB_AGENT_EXTRA_ARGS": "--max-submits-per-run 3", + }, + clear=False, + ): + command = ENTRYPOINT._worker_command() + + self.assertEqual(["--max-submits-per-run", "0"], command[-2:]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 3df411a2..06415114 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -5,7 +5,7 @@ import tempfile import threading import unittest from concurrent.futures import ThreadPoolExecutor -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path @@ -20,6 +20,7 @@ from modelhub_client import ModelHubAPIError, ModelHubClientPool, is_duplicate_s from models import HFModelSummary, ModelInspection # noqa: E402 from outcome_tracker import OutcomeTracker # noqa: E402 from submission_claims import SubmissionClaimStore, candidate_key # noqa: E402 +from template_selector import TemplateSelector # noqa: E402 class FakeClient: @@ -154,6 +155,50 @@ class AutoStrategyClient: return {"code": 0, "data": {"id": f"strategy-{self.calls}"}} +class ExhaustedRecentDiscovery: + def __init__(self) -> None: + self.calls: list[object] = [] + self.recent = HFModelSummary( + repo_id="owner/recent-exhausted", + downloads=100, + last_modified=datetime(2026, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + self.older = [ + HFModelSummary( + repo_id=f"owner/older-{index}", + downloads=100, + last_modified=datetime(2025, 1, 1, tzinfo=timezone.utc), + pipeline_tag="text-generation", + ) + for index in range(3) + ] + + def list_recent_models(self, *, updated_after=None, **_kwargs): # noqa: ANN003 + self.calls.append(updated_after) + return [self.recent] if updated_after is not None else [self.recent, *self.older] + + @staticmethod + def inspect_model(model: HFModelSummary) -> ModelInspection: + return ModelInspection(repo_id=model.repo_id, weight_files=["model.safetensors"]) + + +class ExhaustedRecentClient(AutoStrategyClient): + def __init__(self, available: int, processed_gpus: set[str]) -> None: + super().__init__(available) + self.processed_gpus = processed_gpus + + def processed_gpus_for_model(self, model_id: str) -> set[str]: + if model_id == "owner/recent-exhausted": + return set(self.processed_gpus) + return set() + + def add_task(self, _payload: dict) -> dict: + self.calls += 1 + self.available -= 1 + return {"code": 0, "data": {"id": f"fallback-{self.calls}"}} + + def make_candidate(index: int) -> dict: return { "repoId": f"owner/model-{index}", @@ -163,6 +208,47 @@ def make_candidate(index: int) -> dict: class ClientPoolConcurrencyTests(unittest.TestCase): + def test_exhausted_recent_window_expands_to_older_models_in_same_run(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + args = build_parser().parse_args( + [ + "--task-types", + "text-generation", + "--since-hours", + "48", + "--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") + args.gpu_strategy_state_path = str(root / "strategy.json") + selector = TemplateSelector() + supported_gpus = set(selector.supported_target_gpus("text-generation", auto_only=True)) + discovery = ExhaustedRecentDiscovery() + client = ExhaustedRecentClient(available=2, processed_gpus=supported_gpus) + + summary = run_submission( + args, + now=datetime(2026, 1, 2, 12, tzinfo=timezone.utc), + hf_discovery=discovery, # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + template_selector=selector, + ) + + self.assertEqual(2, summary["submittedCount"]) + self.assertEqual(len(supported_gpus), summary["skipReasonCounts"]["already_processed_for_gpu"]) + self.assertEqual("all_history", summary["scanStages"][-1]["name"]) + self.assertIsNone(discovery.calls[-1]) + def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: root = Path(temporary_dir) @@ -305,7 +391,7 @@ class ClientPoolConcurrencyTests(unittest.TestCase): self.assertEqual(0, len(full_elsewhere.submitted)) self.assertEqual(1, len(available.submitted)) - def test_verification_cache_is_reset_between_cycles(self) -> None: + def test_verification_cache_is_reused_between_cycles(self) -> None: client = FakeClient() pool = ModelHubClientPool([client], active_task_cap=2) # type: ignore[arg-type] pool.search_by_model_id("owner/model") @@ -313,7 +399,7 @@ class ClientPoolConcurrencyTests(unittest.TestCase): self.assertEqual(1, client.search_calls) pool.begin_cycle() pool.search_by_model_id("owner/model") - self.assertEqual(2, client.search_calls) + self.assertEqual(1, client.search_calls) def test_capacity_probe_discovers_a_higher_dynamic_account_limit(self) -> None: client = DynamicCapacityClient(active=2, limit=3) @@ -356,6 +442,31 @@ class ClientPoolConcurrencyTests(unittest.TestCase): class ProcessCoordinationTests(unittest.TestCase): + def test_local_failure_block_expires_after_24_hours(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + tracker = OutcomeTracker(Path(temporary_dir) / "outcomes.jsonl") + now = datetime(2026, 1, 2, 12, tzinfo=timezone.utc) + tracker._records = [ + { + "modelId": "owner/old-failure", + "targetGpu": "gpu-a", + "outcome": "failed", + "submitTime": (now - timedelta(hours=26)).isoformat(), + "lastSyncTime": (now - timedelta(hours=25)).isoformat(), + }, + { + "modelId": "owner/recent-failure", + "targetGpu": "gpu-a", + "outcome": "failed", + "submitTime": (now - timedelta(hours=2)).isoformat(), + "lastSyncTime": (now - timedelta(hours=1)).isoformat(), + }, + ] + tracker._rebuild_indexes() + + self.assertFalse(tracker.is_model_gpu_failed("owner/old-failure", "gpu-a", now=now)) + self.assertTrue(tracker.is_model_gpu_failed("owner/recent-failure", "gpu-a", now=now)) + 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"