Bootstrap architecture rules from task history

modelhub_submmit_api/poll_runner.py

@@ -5,6 +5,8 @@ import json
 import os
 import sys
 import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import timedelta
 from pathlib import Path
 from typing import Any, Callable
 
@@ -36,6 +38,14 @@ DEFAULT_ARCHITECTURE_BLACKLIST_PATH = Path(
 )
 
 
+def _env_int(name: str, default: int, *, minimum: int = 0, maximum: int = 100_000) -> int:
+    try:
+        value = int(os.getenv(name, str(default)))
+    except ValueError:
+        value = default
+    return min(maximum, max(minimum, value))
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Continuously poll ModelHub queue slots and refill submissions.")
     parser.add_argument("--daily-target", type=int, default=0, help="Total submissions to aim for per UTC day; 0 means unlimited")
@@ -309,6 +319,175 @@ def _load_task_compatibility_contexts(
     return contexts
 
 
+def _load_complete_owned_history(
+    modelhub_client: ModelHubClient | ModelHubClientPool,
+) -> tuple[list[dict[str, Any]], list[int]]:
+    clients = (
+        list(modelhub_client.clients)
+        if isinstance(modelhub_client, ModelHubClientPool)
+        else [modelhub_client]
+    )
+    by_account: dict[int, list[dict[str, Any]]] = {}
+    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
+            for index, client in enumerate(clients, start=1)
+        }
+        for future in as_completed(futures):
+            account_index = futures[future]
+            try:
+                by_account[account_index] = future.result()
+            except Exception:
+                errors.append(account_index)
+
+    deduped: dict[str, dict[str, Any]] = {}
+    anonymous: list[dict[str, Any]] = []
+    for account_index in sorted(by_account):
+        for task in by_account[account_index]:
+            if not isinstance(task, dict):
+                continue
+            task_id = task.get("taskId")
+            if task_id is None:
+                anonymous.append(task)
+                continue
+            deduped[str(task_id)] = task
+    return [*deduped.values(), *anonymous], sorted(errors)
+
+
+def _bootstrap_architecture_history(
+    *,
+    modelhub_client: ModelHubClient | ModelHubClientPool,
+    outcome_tracker: OutcomeTracker,
+    ledger_path: Path,
+    now,
+) -> dict[str, Any]:
+    """Prefer public failure evidence, then fall back to all owned history."""
+    probe_size = _env_int(
+        "MODELHUB_ARCHITECTURE_COMMUNITY_PROBE_SIZE",
+        50,
+        minimum=1,
+        maximum=100,
+    )
+    community_records: list[dict[str, Any]] = []
+    community_probe_error: str | None = None
+    try:
+        probe_payload = modelhub_client.list_tasks_page(
+            current=1,
+            page_size=probe_size,
+            only_mine=False,
+            status="success",
+            verify_result=-1,
+        )
+        probe_records = (probe_payload.get("data") or {}).get("records") or []
+        community_records = [record for record in probe_records if isinstance(record, dict)]
+    except Exception as exc:
+        community_probe_error = f"{type(exc).__name__}: {exc}"
+
+    community_usable = [
+        record
+        for record in community_records
+        if record.get("logCosUrl")
+        and record.get("modelId")
+        and record.get("gpuType")
+    ]
+    log(
+        f"[architecture-bootstrap] community_probe={len(community_records)} "
+        f"usable_failure_details={len(community_usable)} "
+        f"error={'none' if community_probe_error is None else community_probe_error}"
+    )
+
+    source = "community_latest"
+    listing_errors: list[int] = []
+    if community_usable:
+        lookback_days = _env_int(
+            "MODELHUB_ARCHITECTURE_COMMUNITY_LOOKBACK_DAYS",
+            30,
+            minimum=1,
+            maximum=365,
+        )
+        latest_limit = _env_int(
+            "MODELHUB_ARCHITECTURE_COMMUNITY_LATEST_LIMIT",
+            5000,
+            minimum=1,
+            maximum=50_000,
+        )
+        try:
+            history_tasks = modelhub_client.list_tasks(
+                page_size=100,
+                only_mine=False,
+                begin_time=now - timedelta(days=lookback_days),
+                end_time=now,
+                status="success",
+                verify_result=-1,
+                max_records=latest_limit,
+            )
+            history_tasks = [task for task in history_tasks if task.get("logCosUrl")]
+            log(
+                f"[architecture-bootstrap] source=community_latest "
+                f"lookback_days={lookback_days} records={len(history_tasks)} limit={latest_limit}"
+            )
+        except Exception as exc:
+            source = "owned_full_history"
+            log(
+                f"[architecture-bootstrap] community_history_error={type(exc).__name__}: {exc} "
+                "fallback=owned_full_history"
+            )
+            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)
+        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'}"
+        )
+
+    contexts = _load_task_compatibility_contexts(
+        outcome_tracker,
+        ledger_path=ledger_path,
+    )
+    summary = outcome_tracker.bootstrap_from_history_tasks(
+        history_tasks,
+        task_contexts=contexts,
+        enrichment_limit=_env_int(
+            "MODELHUB_ARCHITECTURE_BOOTSTRAP_MAX_LOGS",
+            0,
+            minimum=0,
+            maximum=100_000,
+        ),
+        enrichment_workers=_env_int(
+            "MODELHUB_ARCHITECTURE_BOOTSTRAP_WORKERS",
+            8,
+            minimum=1,
+            maximum=16,
+        ),
+        log=log,
+    )
+    feedback = outcome_tracker.get_stats_report()
+    block_count = len(feedback.get("architectureCompatibilityBlocks") or {})
+    summary.update(
+        {
+            "source": source,
+            "communityProbeRecords": len(community_records),
+            "communityUsableFailureDetails": len(community_usable),
+            "listingErrorAccounts": listing_errors,
+            "architectureBlocks": block_count,
+        }
+    )
+    log(
+        f"[architecture-bootstrap] finished source={source} "
+        f"terminal={summary['terminalRecords']} failure_logs={summary['eligibleFailureLogs']} "
+        f"explicit_architecture_failures={summary['explicitArchitectureFailures']} "
+        f"recovered_frameworks={summary['recoveredFrameworks']} blocks={block_count}"
+    )
+    return summary
+
+
 def run_poll_loop(
     *,
     base_args: argparse.Namespace,
@@ -338,6 +517,38 @@ def run_poll_loop(
         f"poll_interval={base_args.poll_interval_seconds}s idle_interval={base_args.idle_interval_seconds}s"
     )
 
+    architecture_bootstrap_summary: dict[str, Any] = {"enabled": False}
+    if not getattr(base_args, "skip_outcome_sync", False):
+        try:
+            architecture_bootstrap_summary = _bootstrap_architecture_history(
+                modelhub_client=modelhub_client,
+                outcome_tracker=outcome_tracker,
+                ledger_path=Path(base_args.ledger_path),
+                now=now,
+            )
+            architecture_bootstrap_summary["enabled"] = True
+            _persist_architecture_blacklist(
+                outcome_tracker.get_stats_report(),
+                path=Path(
+                    getattr(
+                        base_args,
+                        "architecture_blacklist_path",
+                        DEFAULT_ARCHITECTURE_BLACKLIST_PATH,
+                    )
+                ),
+            )
+        except Exception as exc:
+            architecture_bootstrap_summary = {
+                "enabled": True,
+                "error": f"{type(exc).__name__}: {exc}",
+            }
+            log(
+                f"[architecture-bootstrap] error={type(exc).__name__}: {exc} "
+                "continue_polling=true"
+            )
+    else:
+        architecture_bootstrap_summary["reason"] = "outcome_sync_disabled"
+
     cycle_summaries: list[dict[str, Any]] = []
     queue_cleanup_runs: list[dict[str, Any]] = []
     submitted_total = 0
@@ -608,6 +819,7 @@ def run_poll_loop(
         "pollRunDir": str(poll_run_dir),
         "cycleSummaries": cycle_summaries,
         "queueCleanupRuns": queue_cleanup_runs,
+        "architectureBootstrap": architecture_bootstrap_summary,
         "outcomeStats": stats_report,
     }
     write_json(poll_run_dir / "summary.json", summary)