CoolBoy/submmit
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

fix: stream cold-start history into decision state

Browse Source
This commit is contained in:
CoolBoy
2026-09-04 11:20:51 +08:00
parent d60551e130
commit 54ed4351a0
7 changed files with 538 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

362
modelhub_submmit_api/poll_runner.py
View File

@@ -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: