diff --git a/README.md b/README.md index 4926643a..b24fdd42 100644 --- a/README.md +++ b/README.md @@ -61,9 +61,13 @@ Optional tuning: - `MODELHUB_ARCHITECTURE_BLOCK_TTL_DAYS` default `30` - `MODELHUB_ARCHITECTURE_COMMUNITY_PROBE_SIZE` default `50` - `MODELHUB_ARCHITECTURE_COMMUNITY_LOOKBACK_DAYS` default `30` -- `MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT` default `5000` +- `MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT` default `100` (synchronous seed only) - `MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS` default `8` -- `MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS` default `0` (unlimited) +- `MODELHUB_ARCHITECTURE_BOOTSTRAP_TASKS_PER_ACCOUNT` default `10` (synchronous seed only) +- `MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS` default `120` (synchronous seed only) +- `MODELHUB_ARCHITECTURE_BACKFILL_PAGE_SIZE` default `50` +- `MODELHUB_ARCHITECTURE_BACKFILL_PAGES_PER_CYCLE` default `2` +- `MODELHUB_ARCHITECTURE_BACKFILL_LOGS_PER_CYCLE` default `100` - `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` default `5` per account - `MODELHUB_RECENT_MODEL_DAYS` default `7` - `MODELHUB_STATE_SYNC_REMOTE` default `https://dev.modelhub.org.cn/CoolBoy/submmit.git` @@ -145,9 +149,10 @@ At startup the worker verifies the manifest and file checksums, restores local capacity, outcomes, routing evidence, architecture rules, exclusions, intents, and active-task context, then reconciles every account against the platform. Pending intents are held for two hours before being released as unconfirmed. -The branch keeps 30 days of structured events plus lifetime aggregate counters; -raw stdout, credentials, request headers, downloaded archives, and full configs -are never copied. Git authentication uses a temporary `GIT_ASKPASS` helper, so +The branch keeps bounded recovery rows plus lifetime aggregate counters; +raw stdout, credentials, request headers, downloaded archives, complete task +history, and full configs are never copied. Git authentication uses an in-memory +HTTP authentication callback, so the password is absent from command arguments, remotes, commits, and logs. `GET /health` reports process liveness. `GET /ready` returns HTTP 200 only after @@ -168,17 +173,17 @@ change that window. The stats report exposes `architectureCompatibilityBlocks` and per-GPU/framework block counts. The live snapshot is written to `.modelhub_state/architecture_compatibility_blacklist.json`. -Before the first cleanup/submission cycle, the poller probes the latest public -failed-validation records for usable failure archives. If the public endpoint -does not expose those archives (the current API behavior), it falls back to a -parallel, complete history scan of every configured account. Ledger metadata is -used when present; otherwise the task type is recovered from ModelHub's task -level and the selected framework is conservatively recovered from the target -container image in the failure archive. All usable historical failures are -classified once at startup and cached in `outcomes/submissions.jsonl`; subsequent -poll cycles continue incremental learning every three cycles. Empty or -incomplete public evidence is logged as a fallback, never as proof that no -architecture incompatibility exists. +Before the first cleanup/submission cycle, the poller imports only a small recent +seed so cold start cannot be dominated by historical API and log downloads. The +submission pass then runs first on every cycle. Afterwards a resumable backfill +walks every configured account's complete history in bounded pages, classifies +each available failure log, and immediately reduces the page to cumulative +success/failure statistics, memory observations, safe configurations, and exact +compatibility blocks. Only the cursor, temporary deduplication task IDs, and +derived decision checkpoint are synchronized; raw task rows and logs are not. +The task-ID set is deleted when backfill completes. A restart resumes the cursor +instead of rescanning prior pages. Empty public evidence is a fallback signal, +never proof that no incompatibility exists. Before a candidate reaches the submit queue, failure-informed preflight checks the actual ModelScope repository structure and file sizes. Non-GGUF text @@ -425,11 +430,20 @@ those windows are discarded after their category, compatibility block, memory observation, or safe configuration vector has been extracted. Existing archive branches are left untouched but are no longer read or updated. +Version `2026.09.04.3` makes cold recovery incremental and exhaustive. A missing +state branch no longer triggers a blocking all-account/all-log scan. A small +synchronous seed creates the first decision checkpoint, then each submission +cycle consumes two 50-row history pages and up to 100 failure logs after the +submission pass. Progress is durable across restarts, all pages are eventually +processed, and each page is compacted immediately into decision state. It also +fixes large-batch classification so checkpoint compaction cannot discard an +uninspected failure row. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-v31 -m "ModelHub agent 2026.09.04.2" -git push origin main agent-v31 +git tag -a agent-v32 -m "ModelHub agent 2026.09.04.3" +git push origin main agent-v32 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 12e7d7e0..257da3ec 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -103,13 +103,12 @@ bash run_poll.sh --dry-run rule immediately launches a lightweight architecture-only queue scan. Exact matching waiting tasks are stopped after two state checks; running tasks and tasks without local framework/task metadata are protected. -- Startup probes recent community failures for downloadable diagnostic archives. - Because the current public task API omits them, the worker automatically scans - complete terminal history across all configured accounts instead. It joins - ledger context when available and otherwise recovers task type and conservative - framework evidence from the ModelHub task level and target container image. - This bootstrap finishes before the first queue cleanup; later failures continue - to update the same blacklist every three cycles. +- Startup imports a small community/owned seed for immediate decisions. After + each submission pass, a durable cursor incrementally scans complete terminal + history across all configured accounts. Each bounded page is classified and + folded into aggregate decision state immediately; raw task rows and failure + logs are never synchronized. Restarts resume the cursor, and the temporary + deduplication IDs are removed when the exhaustive backfill completes. - A strategy generation lasts exactly 200 platform-accepted submissions. Rejected API calls and 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 @@ -244,6 +243,7 @@ Persistent local scheduler state is written under `.modelhub_state/`: - `submission_exclusions.jsonl`: non-retryable model/GPU uniqueness rejections - `queue_cleanup_latest.json`: latest active-task sizing evidence and cancellation result - `architecture_compatibility_blacklist.json`: current dynamic compatibility blocks and evidence +- `architecture_history_backfill.json`: resumable full-history cursor and temporary deduplication IDs ## Verification diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index c99cb3c5..ac48ffc8 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -537,31 +537,33 @@ class OutcomeTracker: < FAILURE_ENRICHMENT_MAX_ATTEMPTS ) ] - candidates.sort(key=_outcome_record_timestamp, reverse=True) + # The current streamed page must be classified before it is folded + # into the checkpoint. Older retryable rows are secondary; otherwise + # they could consume the batch budget and make the backfill forget a + # newly imported failure without ever inspecting its log. + candidates.sort( + key=lambda record: ( + str(record.get("taskId") or "") in seen_task_ids, + _outcome_record_timestamp(record), + ), + reverse=True, + ) if enrichment_limit > 0: candidates = candidates[:enrichment_limit] self._last_sync_time = utc_now() self._rebuild_failed_index() - if imported or refreshed: - self.save() - candidate_task_ids = [ - str(record["taskId"]) - for record in candidates - if record.get("taskId") is not None - ] enrichment_attempts = 0 classified_total = 0 errors_total = 0 batch_size = 200 - total_candidates = len(candidate_task_ids) + total_candidates = len(candidates) for offset in range(0, total_candidates, batch_size): - batch_ids = candidate_task_ids[offset : offset + batch_size] - batch_records = [ - self._by_task_id[task_id] - for task_id in batch_ids - if task_id in self._by_task_id - ] + # Keep direct references until classification completes. Calling + # save() may compact terminal rows into the aggregate checkpoint; + # looking them up by task ID afterwards would silently drop older + # failure logs from a large import batch. + batch_records = candidates[offset : offset + batch_size] def progress( completed: int, @@ -590,6 +592,8 @@ class OutcomeTracker: errors_total += sum( 1 for record in batch_records if record.get("failureEnrichmentError") ) + + if imported or refreshed or candidates: self._rebuild_failed_index() self.save() @@ -612,6 +616,11 @@ class OutcomeTracker: "recoveredFrameworks": recovered_frameworks, } + def compact_decision_state(self) -> bool: + """Fold terminal rows into the bounded aggregate checkpoint now.""" + self.save() + return self._compact_if_needed(force=True) + @staticmethod def _enrich_history_task( task: dict[str, Any], diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 5f57c320..c134c700 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -11,7 +11,7 @@ from datetime import timedelta from pathlib import Path from typing import Any, Callable -from common import read_jsonl, utc_now, write_json +from common import read_json, read_jsonl, utc_now, write_json from daily_runner import DEFAULT_DAILY_RUNS_DIR, log, run_daily_batches from gpu_strategy import DEFAULT_GPU_STRATEGY_PATH from hf_discovery import HuggingFaceDiscovery @@ -48,6 +48,10 @@ DEFAULT_POLL_RUNS_DIR = Path("poll_runs") DEFAULT_ARCHITECTURE_BLACKLIST_PATH = Path( ".modelhub_state/architecture_compatibility_blacklist.json" ) +DEFAULT_ARCHITECTURE_BACKFILL_PATH = Path( + ".modelhub_state/architecture_history_backfill.json" +) +ARCHITECTURE_BACKFILL_VERSION = 1 def _env_int(name: str, default: int, *, minimum: int = 0, maximum: int = 100_000) -> int: @@ -330,8 +334,10 @@ def _load_task_compatibility_contexts( return contexts -def _load_complete_owned_history( +def _load_bounded_owned_history( modelhub_client: ModelHubClient | ModelHubClientPool, + *, + max_records_per_account: int, ) -> tuple[list[dict[str, Any]], list[int]]: clients = ( list(modelhub_client.clients) @@ -342,7 +348,12 @@ def _load_complete_owned_history( errors: list[int] = [] with ThreadPoolExecutor(max_workers=min(12, max(1, len(clients)))) as executor: futures = { - executor.submit(client.list_tasks, page_size=100, only_mine=True): index + executor.submit( + client.list_tasks, + page_size=100, + only_mine=True, + max_records=max_records_per_account, + ): index for index, client in enumerate(clients, start=1) } for future in as_completed(futures): @@ -373,7 +384,7 @@ def _bootstrap_architecture_history( ledger_path: Path, now, ) -> dict[str, Any]: - """Prefer public failure evidence, then fall back to all owned history.""" + """Prefer bounded recent public evidence, then bounded owned history.""" probe_size = _env_int( "MODELHUB_ARCHITECTURE_COMMUNITY_PROBE_SIZE", 50, @@ -419,7 +430,7 @@ def _bootstrap_architecture_history( ) latest_limit = _env_int( "MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT", - 5000, + 100, minimum=1, maximum=50_000, ) @@ -439,23 +450,41 @@ def _bootstrap_architecture_history( f"lookback_days={lookback_days} records={len(history_tasks)} limit={latest_limit}" ) except Exception as exc: - source = "owned_full_history" + source = "owned_recent_bounded" log( f"[architecture-bootstrap] community_history_error={type(exc).__name__}: {exc} " - "fallback=owned_full_history" + "fallback=owned_recent_bounded" + ) + history_tasks, listing_errors = _load_bounded_owned_history( + modelhub_client, + max_records_per_account=_env_int( + "MODELHUB_ARCHITECTURE_BOOTSTRAP_TASKS_PER_ACCOUNT", + 10, + minimum=10, + maximum=500, + ), ) - history_tasks, listing_errors = _load_complete_owned_history(modelhub_client) else: - source = "owned_full_history" - history_tasks, listing_errors = _load_complete_owned_history(modelhub_client) + source = "owned_recent_bounded" + per_account_limit = _env_int( + "MODELHUB_ARCHITECTURE_BOOTSTRAP_TASKS_PER_ACCOUNT", + 10, + minimum=10, + maximum=500, + ) + history_tasks, listing_errors = _load_bounded_owned_history( + modelhub_client, + max_records_per_account=per_account_limit, + ) account_count = ( len(modelhub_client.clients) if isinstance(modelhub_client, ModelHubClientPool) else 1 ) log( - f"[architecture-bootstrap] source=owned_full_history records={len(history_tasks)} " - f"accounts={account_count} listing_errors={','.join(map(str, listing_errors)) or 'none'}" + f"[architecture-bootstrap] source=owned_recent_bounded records={len(history_tasks)} " + f"accounts={account_count} per_account_limit={per_account_limit} " + f"listing_errors={','.join(map(str, listing_errors)) or 'none'}" ) contexts = _load_task_compatibility_contexts( @@ -467,7 +496,7 @@ def _bootstrap_architecture_history( task_contexts=contexts, enrichment_limit=_env_int( "MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS", - 0, + 120, minimum=0, maximum=100_000, ), @@ -499,6 +528,231 @@ def _bootstrap_architecture_history( return summary +def _architecture_backfill_clients( + modelhub_client: ModelHubClient | ModelHubClientPool, +) -> list[ModelHubClient]: + if isinstance(modelhub_client, ModelHubClientPool): + return list(modelhub_client.clients) + return [modelhub_client] + + +def _new_architecture_backfill_progress( + modelhub_client: ModelHubClient | ModelHubClientPool, + *, + now, +) -> dict[str, Any]: + clients = _architecture_backfill_clients(modelhub_client) + return { + "version": ARCHITECTURE_BACKFILL_VERSION, + "mode": "incremental_decision_only", + "startedAt": now.isoformat(), + # Freeze the upper edge so new submissions cannot continuously move + # historical pagination while the one-time backfill is in progress. + "cutoffAt": now.isoformat(), + "updatedAt": now.isoformat(), + "complete": False, + "nextAccountIndex": 0, + "accounts": { + str(index): { + "nextPage": 1, + "complete": False, + "recordsScanned": 0, + "uniqueRecords": 0, + "listingErrors": 0, + } + for index in range(len(clients)) + }, + # IDs are temporary cursor integrity data, not raw task/log history. + # They are removed as soon as the backfill finishes. + "seenTaskIds": [], + "recordsScanned": 0, + "uniqueRecords": 0, + "terminalRecords": 0, + "failureLogsInspected": 0, + "architectureBlocks": 0, + } + + +def _load_architecture_backfill_progress( + path: Path, + modelhub_client: ModelHubClient | ModelHubClientPool, + *, + now, +) -> dict[str, Any]: + try: + progress = read_json(path) + except (FileNotFoundError, ValueError, TypeError): + progress = {} + clients = _architecture_backfill_clients(modelhub_client) + if ( + not isinstance(progress, dict) + or int(progress.get("version") or 0) != ARCHITECTURE_BACKFILL_VERSION + or not isinstance(progress.get("accounts"), dict) + or len(progress.get("accounts") or {}) != len(clients) + ): + progress = _new_architecture_backfill_progress(modelhub_client, now=now) + return progress + + +def _advance_architecture_history_backfill( + *, + modelhub_client: ModelHubClient | ModelHubClientPool, + outcome_tracker: OutcomeTracker, + ledger_path: Path, + progress_path: Path = DEFAULT_ARCHITECTURE_BACKFILL_PATH, + now=None, +) -> dict[str, Any]: + """Consume a bounded history slice and persist only derived decisions. + + The temporary page cursor and task-ID set make the scan resumable and + prevent page movement from double-counting. Raw task rows and downloaded + failure logs never enter the Git state snapshot. + """ + now = now or utc_now() + clients = _architecture_backfill_clients(modelhub_client) + progress = _load_architecture_backfill_progress( + progress_path, + modelhub_client, + now=now, + ) + if bool(progress.get("complete")): + return progress + + page_size = _env_int( + "MODELHUB_ARCHITECTURE_BACKFILL_PAGE_SIZE", + 50, + minimum=10, + maximum=100, + ) + pages_per_cycle = _env_int( + "MODELHUB_ARCHITECTURE_BACKFILL_PAGES_PER_CYCLE", + 2, + minimum=1, + maximum=12, + ) + max_logs = _env_int( + "MODELHUB_ARCHITECTURE_BACKFILL_LOGS_PER_CYCLE", + page_size * pages_per_cycle, + minimum=1, + maximum=500, + ) + accounts = progress.get("accounts") or {} + seen_ids = {str(value) for value in (progress.get("seenTaskIds") or []) if value is not None} + batch_tasks: list[dict[str, Any]] = [] + page_calls = 0 + page_errors = 0 + + for _ in range(pages_per_cycle): + incomplete = [ + index + for index in range(len(clients)) + if not bool((accounts.get(str(index)) or {}).get("complete")) + ] + if not incomplete: + break + start = int(progress.get("nextAccountIndex") or 0) % max(1, len(clients)) + account_index = next( + (index for index in range(start, len(clients)) if index in incomplete), + incomplete[0], + ) + state = accounts[str(account_index)] + current_page = max(1, int(state.get("nextPage") or 1)) + progress["nextAccountIndex"] = (account_index + 1) % len(clients) + page_calls += 1 + try: + payload = clients[account_index].list_tasks_page( + current=current_page, + page_size=page_size, + only_mine=True, + end_time=str(progress.get("cutoffAt") or now.isoformat()), + ) + data = payload.get("data") or {} + records = [item for item in (data.get("records") or []) if isinstance(item, dict)] + pages = max(0, int(data.get("pages") or 0)) + state["recordsScanned"] = int(state.get("recordsScanned") or 0) + len(records) + progress["recordsScanned"] = int(progress.get("recordsScanned") or 0) + len(records) + new_count = 0 + for task in records: + task_id = task.get("taskId") + if task_id is None: + batch_tasks.append(task) + new_count += 1 + continue + key = str(task_id) + if key in seen_ids: + continue + seen_ids.add(key) + batch_tasks.append(task) + new_count += 1 + state["uniqueRecords"] = int(state.get("uniqueRecords") or 0) + new_count + progress["uniqueRecords"] = int(progress.get("uniqueRecords") or 0) + new_count + if not records or pages <= current_page: + state["complete"] = True + state["completedAt"] = now.isoformat() + else: + state["nextPage"] = current_page + 1 + except Exception as exc: + page_errors += 1 + state["listingErrors"] = int(state.get("listingErrors") or 0) + 1 + state["lastError"] = f"{type(exc).__name__}: {exc}"[:500] + + batch_summary: dict[str, Any] = { + "terminalRecords": 0, + "enrichmentAttempts": 0, + "explicitArchitectureFailures": 0, + } + if batch_tasks: + batch_summary = outcome_tracker.bootstrap_from_history_tasks( + batch_tasks, + task_contexts=_load_task_compatibility_contexts( + outcome_tracker, + ledger_path=ledger_path, + ), + enrichment_limit=max_logs, + enrichment_workers=_env_int( + "MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS", + 8, + minimum=1, + maximum=16, + ), + log=log, + ) + # Each slice is immediately reduced to cumulative counters, routing + # statistics, compatibility blocks and a bounded recent window. + outcome_tracker.compact_decision_state() + progress["terminalRecords"] = int(progress.get("terminalRecords") or 0) + int( + batch_summary.get("terminalRecords") or 0 + ) + progress["failureLogsInspected"] = int(progress.get("failureLogsInspected") or 0) + int( + batch_summary.get("enrichmentAttempts") or 0 + ) + progress["architectureBlocks"] = len( + outcome_tracker.get_stats_report().get("architectureCompatibilityBlocks") or {} + ) + + progress["updatedAt"] = now.isoformat() + complete = all(bool((accounts.get(str(index)) or {}).get("complete")) for index in range(len(clients))) + progress["complete"] = complete + if complete: + progress["completedAt"] = now.isoformat() + progress.pop("seenTaskIds", None) + else: + progress["seenTaskIds"] = sorted(seen_ids) + write_json(progress_path, progress) + completed_accounts = sum( + 1 for index in range(len(clients)) if bool((accounts.get(str(index)) or {}).get("complete")) + ) + log( + f"[architecture-backfill] status={'complete' if complete else 'running'} " + f"pages={page_calls} page_errors={page_errors} batch_records={len(batch_tasks)} " + f"failure_logs={int(batch_summary.get('enrichmentAttempts') or 0)} " + f"accounts={completed_accounts}/{len(clients)} " + f"unique_total={int(progress.get('uniqueRecords') or 0)} " + "retention=decision_state_only" + ) + return progress + + def run_poll_loop( *, base_args: argparse.Namespace, @@ -585,21 +839,33 @@ def run_poll_loop( ) architecture_bootstrap_summary: dict[str, Any] = {"enabled": False} + architecture_backfill_path = DEFAULT_ARCHITECTURE_BACKFILL_PATH + had_durable_checkpoint = outcome_tracker.has_durable_checkpoint if not getattr(base_args, "skip_outcome_sync", False): try: - if outcome_tracker.has_durable_checkpoint: + if had_durable_checkpoint: cached_feedback = outcome_tracker.get_stats_report() + try: + restored_backfill = read_json(architecture_backfill_path) + except (FileNotFoundError, ValueError, TypeError): + restored_backfill = {} + backfill_pending = bool( + isinstance(restored_backfill, dict) + and restored_backfill + and not bool(restored_backfill.get("complete")) + ) architecture_bootstrap_summary = { "source": "durable_checkpoint", "terminalRecords": int(cached_feedback.get("terminalRecords") or 0), "architectureBlocks": len(cached_feedback.get("architectureCompatibilityBlocks") or {}), - "fullHistoryScanSkipped": True, + "fullHistoryScanSkipped": not backfill_pending, + "historyBackfill": "resuming" if backfill_pending else "complete_or_legacy", } log( "[architecture-bootstrap] source=durable_checkpoint " f"terminal={architecture_bootstrap_summary['terminalRecords']} " f"blocks={architecture_bootstrap_summary['architectureBlocks']} " - "full_history_scan=skipped" + f"history_backfill={architecture_bootstrap_summary['historyBackfill']}" ) else: architecture_bootstrap_summary = _bootstrap_architecture_history( @@ -608,6 +874,18 @@ def run_poll_loop( ledger_path=Path(base_args.ledger_path), now=now, ) + outcome_tracker.compact_decision_state() + # A missing decision checkpoint means the prior aggregate was + # unavailable. Start a resumable full owned-history backfill; + # the small synchronous seed above only makes the first routing + # decisions useful without delaying submissions. + write_json( + architecture_backfill_path, + _new_architecture_backfill_progress( + modelhub_client, + now=now, + ), + ) architecture_bootstrap_summary["enabled"] = True _persist_architecture_blacklist( outcome_tracker.get_stats_report(), @@ -628,6 +906,14 @@ def run_poll_loop( f"[architecture-bootstrap] error={type(exc).__name__}: {exc} " "continue_polling=true" ) + if not had_durable_checkpoint: + write_json( + architecture_backfill_path, + _new_architecture_backfill_progress( + modelhub_client, + now=now, + ), + ) else: architecture_bootstrap_summary["reason"] = "outcome_sync_disabled" @@ -641,6 +927,42 @@ def run_poll_loop( pending_architecture_cleanup = False last_cleaned_architecture_blocks: set[str] = set() + def advance_architecture_backfill() -> None: + nonlocal pending_architecture_cleanup + if getattr(base_args, "skip_outcome_sync", False) or not architecture_backfill_path.is_file(): + return + try: + progress = read_json(architecture_backfill_path) + if isinstance(progress, dict) and bool(progress.get("complete")): + return + previous_blocks = set( + (outcome_tracker.get_stats_report().get("architectureCompatibilityBlocks") or {}).keys() + ) + _advance_architecture_history_backfill( + modelhub_client=modelhub_client, + outcome_tracker=outcome_tracker, + ledger_path=Path(base_args.ledger_path), + progress_path=architecture_backfill_path, + ) + current_feedback = outcome_tracker.get_stats_report() + current_blocks = _persist_architecture_blacklist( + current_feedback, + path=Path( + getattr( + base_args, + "architecture_blacklist_path", + DEFAULT_ARCHITECTURE_BLACKLIST_PATH, + ) + ), + ) + if current_blocks - previous_blocks: + pending_architecture_cleanup = True + except Exception as exc: + log( + f"[architecture-backfill] status=deferred " + f"error={type(exc).__name__}: {exc} continue_polling=true" + ) + while True: if base_args.max_cycles and cycles >= base_args.max_cycles: stopped_reason = "max_cycles_reached" @@ -840,6 +1162,10 @@ def run_poll_loop( ) if available_slots is not None and available_slots <= 0: + # Historical learning is deliberately behind queue maintenance and + # capacity checks. It advances even while full, but never blocks + # the worker's initial recovery or first submission attempt. + advance_architecture_backfill() if state_sync is not None: try: if cycles % 3 == 0 and hasattr(modelhub_client, "list_active_tasks_by_account"): @@ -879,6 +1205,10 @@ def run_poll_loop( f"stop={cycle_summary['stoppedReason']}" ) + # Do this only after the submission pass. Each cycle consumes at most a + # small configured slice and immediately reduces it to decision state. + advance_architecture_backfill() + if state_sync is not None: try: if cycle_summary.get("submittedTotal", 0) > 0 or cycles % 3 == 0: diff --git a/modelhub_submmit_api/state_sync.py b/modelhub_submmit_api/state_sync.py index f952be3f..839721e2 100644 --- a/modelhub_submmit_api/state_sync.py +++ b/modelhub_submmit_api/state_sync.py @@ -49,6 +49,7 @@ STATE_OUTCOME_FIELDS = { STATE_ALLOWLIST = ( ".modelhub_state/account_capacity.json", ".modelhub_state/architecture_compatibility_blacklist.json", + ".modelhub_state/architecture_history_backfill.json", ".modelhub_state/gpu_strategy.json", ".modelhub_state/market_intelligence.json", ".modelhub_state/official_capabilities.json", diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 2aafadef..87232174 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.09.04.2" +AGENT_VERSION = "2026.09.04.3" diff --git a/tests/test_poll_policy.py b/tests/test_poll_policy.py index 57c82ad9..293ccb5c 100644 --- a/tests/test_poll_policy.py +++ b/tests/test_poll_policy.py @@ -6,6 +6,7 @@ import tempfile import unittest from datetime import datetime, timezone from pathlib import Path +from unittest.mock import patch PACKAGE_DIR = Path(__file__).resolve().parents[1] / "modelhub_submmit_api" @@ -15,6 +16,7 @@ sys.path.insert(0, str(PACKAGE_DIR)) from outcome_tracker import OutcomeTracker # noqa: E402 from poll_runner import ( # noqa: E402 + _advance_architecture_history_backfill, _bootstrap_architecture_history, _load_task_compatibility_contexts, build_parser, @@ -65,10 +67,93 @@ class PollPolicyTests(unittest.TestCase): now=datetime.now(timezone.utc), ) - self.assertEqual("owned_full_history", summary["source"]) + self.assertEqual("owned_recent_bounded", summary["source"]) self.assertEqual(1, summary["terminalRecords"]) self.assertEqual(0, summary["communityUsableFailureDetails"]) + def test_architecture_history_backfill_resumes_and_keeps_only_decision_state(self) -> None: + class PagedHistoryClient: + calls: list[int] = [] + + def list_tasks_page(self, **kwargs): # noqa: ANN003, ANN201 + current = int(kwargs["current"]) + self.calls.append(current) + records = { + 1: [ + { + "taskId": "history-1", + "modelId": "owner/model-1", + "gpuType": "gpu-a", + "status": "success", + "verifyResult": 1, + }, + { + "taskId": "history-2", + "modelId": "owner/model-2", + "gpuType": "gpu-a", + "status": "failed", + "verifyResult": -1, + }, + ], + 2: [ + { + "taskId": "history-3", + "modelId": "owner/model-3", + "gpuType": "gpu-b", + "status": "success", + "verifyResult": 1, + } + ], + }[current] + return {"data": {"records": records, "pages": 2}} + + with tempfile.TemporaryDirectory() as temporary_dir, patch.dict( + "os.environ", + {"MODELHUB_ARCHITECTURE_BACKFILL_PAGES_PER_CYCLE": "1"}, + ): + root = Path(temporary_dir) + outcomes = root / "outcomes.jsonl" + checkpoint = root / "checkpoint.json" + recent = root / "recent.jsonl" + progress_path = root / "backfill.json" + tracker = OutcomeTracker( + outcomes, + checkpoint_path=checkpoint, + recent_path=recent, + ) + client = PagedHistoryClient() + first = _advance_architecture_history_backfill( + modelhub_client=client, # type: ignore[arg-type] + outcome_tracker=tracker, + ledger_path=root / "ledger.jsonl", + progress_path=progress_path, + now=datetime.now(timezone.utc), + ) + self.assertFalse(first["complete"]) + self.assertTrue(tracker.has_durable_checkpoint) + + restored = OutcomeTracker( + outcomes, + checkpoint_path=checkpoint, + recent_path=recent, + ) + second = _advance_architecture_history_backfill( + modelhub_client=client, # type: ignore[arg-type] + outcome_tracker=restored, + ledger_path=root / "ledger.jsonl", + progress_path=progress_path, + now=datetime.now(timezone.utc), + ) + + self.assertTrue(second["complete"]) + self.assertNotIn("seenTaskIds", second) + self.assertEqual([1, 2], client.calls) + self.assertEqual(3, restored.get_stats_report()["terminalRecords"]) + self.assertEqual([], json.loads(outcomes.read_text(encoding="utf-8") or "[]")) + persisted_progress = progress_path.read_text(encoding="utf-8") + self.assertNotIn("owner/model", persisted_progress) + self.assertNotIn("logs.invalid", persisted_progress) + def test_cleanup_contexts_merge_outcomes_with_older_ledger_entries(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: root = Path(temporary_dir) @@ -106,6 +191,47 @@ class PollPolicyTests(unittest.TestCase): self.assertEqual("mindie", contexts["task-old"]["framework"]) self.assertEqual({}, contexts["task-old"]["modelProfile"]) + def test_history_batch_is_classified_before_checkpoint_compaction(self) -> None: + tasks = [ + { + "taskId": f"failed-{index}", + "modelId": f"owner/model-{index}", + "gpuType": "gpu-a", + "status": "failed", + "verifyResult": -1, + "logCosUrl": f"https://logs.invalid/{index}", + } + for index in range(600) + ] + + def classify(records, **kwargs): # noqa: ANN001, ANN003, ANN202 + for record in records: + record["failureCategory"] = "model_runtime" + record["failureScope"] = "model" + record["failureDeterministic"] = True + progress = kwargs.get("progress") + if progress is not None: + progress(len(records), len(records), len(records), 0) + return len(records) + + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + tracker = OutcomeTracker( + root / "outcomes.jsonl", + checkpoint_path=root / "checkpoint.json", + recent_path=root / "recent.jsonl", + ) + with patch.object(tracker, "_enrich_failure_records", side_effect=classify): + summary = tracker.bootstrap_from_history_tasks( + tasks, + enrichment_limit=600, + ) + + report = tracker.get_stats_report() + self.assertEqual(600, summary["enrichmentAttempts"]) + self.assertEqual(600, report["totals"]["attributableFailureCount"]) + self.assertEqual(600, report["totals"]["failureBreakdown"]["model_runtime"]) + def test_age_policy_defaults_to_admission_only_reserve_five(self) -> None: args = build_parser().parse_args([])