From ff737685370fc8c978e8cf0910def0f690577622 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Fri, 4 Sep 2026 10:30:21 +0800 Subject: [PATCH] feat: compact durable logs and refine framework routing --- README.md | 21 ++- docs/failure-analysis-2026-09-04.md | 83 +++++++++ modelhub_submmit_api/main.py | 89 +++++++++- modelhub_submmit_api/market_intelligence.py | 12 +- modelhub_submmit_api/modelhub_client.py | 13 ++ modelhub_submmit_api/official_capabilities.py | 15 ++ modelhub_submmit_api/outcome_tracker.py | 60 ++++++- modelhub_submmit_api/poll_runner.py | 31 +++- modelhub_submmit_api/routing_engine.py | 14 ++ modelhub_submmit_api/state_sync.py | 167 ++++++++++++++---- modelhub_submmit_api/version.py | 2 +- tests/test_concurrency.py | 39 ++++ tests/test_super_agent.py | 119 +++++++++++++ 13 files changed, 621 insertions(+), 44 deletions(-) create mode 100644 docs/failure-analysis-2026-09-04.md diff --git a/README.md b/README.md index c25aa458..c22a9ac3 100644 --- a/README.md +++ b/README.md @@ -396,11 +396,28 @@ the aggregate checkpoint remains the authoritative recovery input even while an archive shard is awaiting upload. ModelHub Git traffic is excluded from proxy routing in the container to avoid deployment-specific Git failures. +Version `2026.09.04.1` makes durable state synchronization content-addressed: +unchanged cycles do not create commits, and a failed network push retries the +same commit/generation. The write-ahead batch size is now 100; resolved intent +rows are compacted to 200 recent entries and moved to monthly gzip archive +branches. Raw community samples and the official model/GPU capability cache are +bounded, so restart no longer needs to restore and rewrite multi-megabyte +disposable caches. + +The submission loop now treats account-capacity saturation and framework +prerequisite responses as deferred policy outcomes rather than model failures. +When a model has an official non-`transformers` route, that immediately +executable route is selected before the platform's prerequisite-gated +`transformers` fallback; a recorded non-transformers failure unlocks normal +success-based transformer selection. Missing framework/task/profile fields in history are +recovered from the durable task ledger. The evidence and category breakdown are +documented in `docs/failure-analysis-2026-09-04.md`. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-v29 -m "ModelHub agent 2026.08.22.1" -git push origin main agent-v29 +git tag -a agent-v30 -m "ModelHub agent 2026.09.04.1" +git push origin main agent-v30 ``` diff --git a/docs/failure-analysis-2026-09-04.md b/docs/failure-analysis-2026-09-04.md new file mode 100644 index 00000000..4b780784 --- /dev/null +++ b/docs/failure-analysis-2026-09-04.md @@ -0,0 +1,83 @@ +# Durable log analysis — 2026-09-04 + +This report was produced from the live `agent-state` snapshot and the monthly +`agent-archive-2026-08` branch. Records were deduplicated by task identity before +classification; intent rows and state-sync generations were not counted as +independent model submissions. + +## Data quality + +- 15,178 raw outcome rows became 14,178 unique outcomes after deduplication. +- 13,318 unique terminal outcomes contained 733 successes and 12,585 failures. +- The hot intent WAL contained 9,213 rows, but only 1,299 unique model/GPU + routes. Capacity-full responses alone appeared 5,887 times. +- The ledger contained 1,028 platform tasks. A single model can legitimately + produce several tasks because GPU/framework routes and retries differ. +- Raw terminal success rate is biased downward: failures often terminate much + faster, while successful tasks can remain waiting/running for a long time. + Routing therefore uses attributable outcomes and confidence bounds rather + than this raw percentage. + +## Deduplicated failure categories + +| Category | Count | Share | +| --- | ---: | ---: | +| Ambiguous runtime | 3,067 | 24.37% | +| Parameters/template | 2,935 | 23.32% | +| Framework/architecture unsupported | 1,428 | 11.35% | +| Memory capacity | 1,169 | 9.29% | +| Platform infrastructure | 911 | 7.24% | +| Repository structure | 725 | 5.76% | +| Missing log evidence | 719 | 5.71% | +| Tokenizer compatibility | 576 | 4.58% | +| Model load | 382 | 3.04% | +| Context length | 318 | 2.53% | +| Architecture compatibility | 212 | 1.68% | +| Runtime memory | 55 | 0.44% | +| Backend operator | 44 | 0.35% | +| Generic validation failure | 43 | 0.34% | +| Attention backend | 1 | 0.01% | + +The most useful structured codes were `MODEL_NOT_SUPPORTED` (1,700), +`EXECUTE_EMPTY_RESULT` (1,392), `PREFLIGHT_OOM` (1,166), `TOKENIZER_FAILED` +(696), `MODEL_FILE_NOT_FOUND` (590), `MODEL_LOAD_FAILED` (382), and +`CONTEXT_LENGTH_ERROR` (268). + +## Representative evidence and algorithm changes + +1. Capacity saturation repeatedly returned the account-limit message. This is + platform flow control, not model/framework failure. It is now recorded as + `capacity_deferred`, ends the current submission cycle after the pool has + tried every account, and does not enter success statistics. +2. The platform repeatedly required a failed non-`transformers` verification + before accepting `transformers`. Routing now uses a vetted non-transformers + route when one exists and treats the prerequisite response as a neutral + deferred result. Once durable history contains a non-transformers failure + for that model, transformers is unlocked and competes normally by success + confidence. +3. `PREFLIGHT_OOM` logs explicitly compare repository load size (including the + platform's 20% loading allowance) with available device memory. These remain + deterministic pre-submit blocks and deterministic queue cleanup evidence. +4. Explicit “framework does not support model/architecture” messages continue + to create exact GPU + framework + task + architecture blocks. Generic image, + driver, storage and connection errors remain platform-neutral and do not + poison compatibility scores. +5. Platform history often omits `framework`. Outcome sync now joins task IDs + with the local durable ledger and restores framework/task/profile metadata, + improving per-framework statistics without guessing from model names. + +## Durable-state corrections + +- Hot intent history retains unresolved intents plus 200 recent terminal + intents; older terminal attempts are gzip archived by month. +- Community raw samples are capped locally at 200 and excluded from the hot Git + snapshot. The aggregated GPU/framework statistics remain durable. +- Model/GPU official-capability cache is bounded to the 1,500 newest entries. +- Outcome compaction starts at 1,000 rows instead of 2,000. +- An unchanged snapshot produces no Git commit. A failed push retries the exact + same commit and generation instead of creating a new generation every minute. +- Submission intent batches default to 100, reducing Git transactions while + preserving write-ahead recovery. + +These changes keep full forensic evidence in cold archive branches while making +the hot branch small enough for quick restart and reliable server-side unpack. diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 5fb4c239..7746646a 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -36,7 +36,9 @@ from modelhub_client import ( ModelHubClient, ModelHubClientPool, OldModelQueuePolicyError, + is_capacity_error, is_duplicate_submission_error, + is_framework_prerequisite_error, is_model_uniqueness_error, ) from models import CandidateModel, HFModelSummary, ModelInspection @@ -651,6 +653,10 @@ def process_model_for_candidates( skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason}) continue record = candidate_to_record(best) + if outcome_tracker is not None: + record["transformersPrerequisiteSatisfied"] = ( + outcome_tracker.has_non_transformers_failure(model.repo_id) + ) if market_intelligence is not None: record.update(market_intelligence.gpu_metadata(target_gpu)) record.update(market_intelligence.framework_metadata(best.task_type, target_gpu, best.framework)) @@ -879,6 +885,17 @@ def submit_candidate( "candidate": candidate, "reason": "age_policy_skipped", } + if is_framework_prerequisite_error(exc): + print( + f"[submit] deferred repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=framework_prerequisite", + flush=True, + ) + return { + "outcome": "framework_prerequisite_deferred", + "candidate": candidate, + "reason": "framework_prerequisite_deferred", + } if is_model_uniqueness_error(exc): print( f"[submit] skipped repo={candidate['repoId']} gpu={candidate['targetGpu']} " @@ -901,6 +918,17 @@ def submit_candidate( "candidate": candidate, "reason": str(exc), } + if is_capacity_error(exc): + print( + f"[submit] deferred repo={candidate['repoId']} gpu={candidate['targetGpu']} " + f"framework={candidate['framework']} reason=account_capacity_saturated", + flush=True, + ) + return { + "outcome": "capacity_deferred", + "candidate": candidate, + "reason": "account_capacity_saturated", + } print( f"[submit] failed repo={candidate['repoId']} gpu={candidate['targetGpu']} " f"framework={candidate['framework']} reason={exc}", @@ -1052,6 +1080,7 @@ def run_submission( run_dir = make_run_dir(runs_dir, now) ledger_path.parent.mkdir(parents=True, exist_ok=True) history_archive_path.parent.mkdir(parents=True, exist_ok=True) + ledger_entries = load_ledger(ledger_path) outcome_tracker = outcome_tracker or OutcomeTracker(Path(args.outcomes_path)) # Online decisions are deliberately deterministic. The optional classifier @@ -1085,7 +1114,25 @@ def run_submission( synced_count = 0 if not getattr(args, "skip_outcome_sync", False): try: - synced_count = outcome_tracker.sync_from_api(modelhub_client) + task_contexts = outcome_tracker.get_task_compatibility_contexts() + for entry in ledger_entries: + task_id = str(entry.get("taskId") or "") + if not task_id: + continue + context = task_contexts.setdefault(task_id, {}) + for field, source in ( + ("modelId", "modelId"), + ("targetGpu", "targetGpu"), + ("framework", "framework"), + ("taskType", "taskType"), + ("submitTime", "submitTime"), + ): + if not context.get(field) and entry.get(source): + context[field] = entry.get(source) + synced_count = outcome_tracker.sync_from_api( + modelhub_client, + task_contexts=task_contexts, + ) except Exception: pass if preflight_advisor is not None: @@ -1096,7 +1143,6 @@ def run_submission( updated_after = determine_updated_after(args, now) history_begin = now - timedelta(days=args.stats_window_days) - ledger_entries = load_ledger(ledger_path) day_start = now.replace(hour=0, minute=0, second=0, microsecond=0) # Count today's submissions from the local ledger (avoids expensive paginated API call) daily_snapshot = count_submissions_for_day(tasks=[], ledger_entries=ledger_entries, day_start=day_start, day_end=now) @@ -1473,6 +1519,7 @@ def run_submission( submit_workers = 1 state_sync = getattr(args, "_state_sync_manager", None) state_sync_paused = False + capacity_saturated = False if args.dry_run: attempted_candidates = diversified_candidates[:target_submit_count] @@ -1549,6 +1596,7 @@ def run_submission( batch_duplicate_candidates: list[dict[str, Any]] = [] batch_uniqueness_rejected_candidates: list[dict[str, Any]] = [] batch_policy_skipped_candidates: list[dict[str, Any]] = [] + batch_capacity_deferred_candidates: list[dict[str, Any]] = [] batch_failed_candidates: list[dict[str, Any]] = [] for index in range(len(batch_candidates)): result = ordered_results.get(index) @@ -1593,6 +1641,30 @@ def run_submission( } ) continue + if result["outcome"] == "capacity_deferred": + capacity_saturated = True + batch_capacity_deferred_candidates.append(candidate) + skipped.append( + { + "repoId": candidate["repoId"], + "targetGpu": candidate["targetGpu"], + "reason": "account_capacity_saturated", + "deferred": True, + } + ) + continue + if result["outcome"] == "framework_prerequisite_deferred": + batch_policy_skipped_candidates.append(candidate) + skipped.append( + { + "repoId": candidate["repoId"], + "targetGpu": candidate["targetGpu"], + "framework": candidate["framework"], + "reason": "framework_prerequisite_deferred", + "deferred": True, + } + ) + continue if result["outcome"] in {"failed", "precheck_deferred"}: batch_failed_candidates.append(candidate) failed.append( @@ -1646,7 +1718,13 @@ def run_submission( claim_store.mark_submitted( [*batch_submitted_candidates, *batch_duplicate_candidates, *batch_uniqueness_rejected_candidates] ) - claim_store.release([*batch_failed_candidates, *batch_policy_skipped_candidates]) + claim_store.release( + [ + *batch_failed_candidates, + *batch_policy_skipped_candidates, + *batch_capacity_deferred_candidates, + ] + ) if strategy_manager is not None: strategy_manager.record_accepted(batch_submitted_candidates) outcome_tracker.save() @@ -1663,6 +1741,11 @@ def run_submission( if hasattr(modelhub_client, "available_submit_slots") and modelhub_client.available_submit_slots() <= 0: break + if capacity_saturated: + # Every account was tried by the pool. More attempts in this + # cycle only duplicate the same capacity response and pollute + # durable logs; refresh capacity on the next poll instead. + break write_jsonl(run_dir / "submitted.jsonl", submitted) write_jsonl(run_dir / "skipped.jsonl", skipped) diff --git a/modelhub_submmit_api/market_intelligence.py b/modelhub_submmit_api/market_intelligence.py index 9ae52597..ede1cbca 100644 --- a/modelhub_submmit_api/market_intelligence.py +++ b/modelhub_submmit_api/market_intelligence.py @@ -22,7 +22,7 @@ DEFAULT_FRAMEWORK_MIN_SAMPLES = 300 DEFAULT_GPU_MIN_RECENT_TERMINALS = 20 DEFAULT_FRAMEWORK_MIN_WILSON = 0.05 DEFAULT_COMMUNITY_WINDOW_DAYS = 30 -DEFAULT_COMMUNITY_MAX_MODELS = 2000 +DEFAULT_COMMUNITY_MAX_MODELS = 200 DEFAULT_COMMUNITY_HYDRATE_PER_REFRESH = 50 NEW_FRAMEWORK_PROMOTION_MARGIN = 1.10 ERROR_RETRY_SECONDS = 300 @@ -248,6 +248,16 @@ class MarketIntelligenceManager: else: state = self._base_state(gpus, tasks, now) + community_sample = state.get("communitySample") or {} + if isinstance(community_sample, dict) and len(community_sample) > DEFAULT_COMMUNITY_MAX_MODELS: + cutoff = now - timedelta(days=DEFAULT_COMMUNITY_WINDOW_DAYS) + ordered = sorted( + community_sample.items(), + key=lambda item: parse_datetime((item[1] or {}).get("updateTime")) or cutoff, + reverse=True, + ) + state["communitySample"] = dict(ordered[:DEFAULT_COMMUNITY_MAX_MODELS]) + queue_due = not _fresh( state.get("queueUpdatedAt"), now=now, diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index a7e534d8..127c05cd 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -370,6 +370,12 @@ MODEL_UNIQUENESS_ERROR_MARKERS = ( "model already exists", ) +FRAMEWORK_PREREQUISITE_ERROR_MARKERS = ( + "必须在非transformers框架验证失败后", + "must fail on a non-transformers framework", + "non-transformers framework verification failed", +) + DEFAULT_CAPACITY_STATE_PATH = Path(".modelhub_state/account_capacity.json") @@ -396,6 +402,13 @@ def is_model_uniqueness_error(error: ModelHubAPIError) -> bool: return any(marker in message for marker in MODEL_UNIQUENESS_ERROR_MARKERS) +def is_framework_prerequisite_error(error: ModelHubAPIError) -> bool: + message = str(error).strip().lower() + if isinstance(error.payload, dict): + message = f"{message} {error.payload.get('message') or ''}".lower() + return any(marker in message for marker in FRAMEWORK_PREREQUISITE_ERROR_MARKERS) + + class ModelHubClientPool: def __init__( self, diff --git a/modelhub_submmit_api/official_capabilities.py b/modelhub_submmit_api/official_capabilities.py index 2f2fc43c..3532a7df 100644 --- a/modelhub_submmit_api/official_capabilities.py +++ b/modelhub_submmit_api/official_capabilities.py @@ -10,6 +10,7 @@ from common import parse_datetime, read_json, utc_now, write_json OFFICIAL_CAPABILITY_VERSION = 1 DEFAULT_OFFICIAL_CAPABILITIES_PATH = Path(".modelhub_state/official_capabilities.json") +DEFAULT_MODEL_GPU_CACHE_LIMIT = 1500 class OfficialCapabilityUnavailable(RuntimeError): @@ -61,6 +62,18 @@ class OfficialCapabilityRegistry: self.pause_reason: str | None = None self._lock = threading.Lock() + @staticmethod + def _prune_model_gpu_cache(state: dict[str, Any]) -> None: + cache = state.get("modelGpuTaskTypes") + if not isinstance(cache, dict) or len(cache) <= DEFAULT_MODEL_GPU_CACHE_LIMIT: + return + ordered = sorted( + cache.items(), + key=lambda pair: parse_datetime((pair[1] or {}).get("updatedAt")) or datetime.min.replace(tzinfo=utc_now().tzinfo), + reverse=True, + ) + state["modelGpuTaskTypes"] = dict(ordered[:DEFAULT_MODEL_GPU_CACHE_LIMIT]) + def _load(self) -> dict[str, Any]: try: value = read_json(self.path) @@ -73,6 +86,7 @@ class OfficialCapabilityRegistry: def prepare(self, client: Any, *, fallback_gpus: list[str], task_types: list[str], now: datetime | None = None) -> dict[str, Any]: now = now or utc_now() state = self._load() + self._prune_model_gpu_cache(state) catalog_usable = _fresh(state.get("catalogUpdatedAt"), now, 1800) task_tree_usable = _fresh(state.get("taskTreeUpdatedAt"), now, 7 * 86400) errors: list[str] = [] @@ -159,6 +173,7 @@ class OfficialCapabilityRegistry: with self._lock: cache = self.state.setdefault("modelGpuTaskTypes", {}) cache[key] = {"updatedAt": now.isoformat(), "taskTypes": task_types} + self._prune_model_gpu_cache(self.state) self.state["generatedAt"] = now.isoformat() write_json(self.path, self.state) return task_types diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index a70ea123..79f79670 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -31,7 +31,7 @@ DEFAULT_OUTCOME_CHECKPOINT_PATH = Path(".modelhub_state/outcome_checkpoint.json" DEFAULT_RECENT_OUTCOMES_PATH = Path(".modelhub_state/recent_outcomes.jsonl") DEFAULT_ARCHIVE_PENDING_DIR = Path(".modelhub_state/archive_pending/outcomes") OUTCOME_CHECKPOINT_VERSION = 1 -DEFAULT_OUTCOME_COMPACT_THRESHOLD = 2000 +DEFAULT_OUTCOME_COMPACT_THRESHOLD = 1000 DEFAULT_RECENT_OUTCOME_LIMIT = 1000 FAILURE_ENRICHMENT_LIMIT = 40 FAILURE_ENRICHMENT_WORKERS = 4 @@ -215,6 +215,25 @@ class OutcomeTracker: } return contexts + def merge_task_contexts(self, contexts: dict[str, dict[str, Any]] | None) -> int: + """Repair framework/profile fields omitted by the platform history API.""" + if not isinstance(contexts, dict): + return 0 + changed = 0 + for task_id, context in contexts.items(): + record = self._by_task_id.get(str(task_id)) + if record is None or not isinstance(context, dict): + continue + before = json.dumps(record, ensure_ascii=False, sort_keys=True) + self._merge_record_metadata(record, self._enrich_history_task({}, context)) + if json.dumps(record, ensure_ascii=False, sort_keys=True) != before: + changed += 1 + if changed: + write_jsonl(self.recent_path, self._recent_records) + self._rebuild_indexes() + self.save() + return changed + def get_strategy_history_records(self) -> list[dict[str, Any]]: """Expose only successes and evidence-attributable failures for GPU ranking.""" records: list[dict[str, Any]] = [] @@ -331,7 +350,12 @@ class OutcomeTracker: self._rebuild_failed_index() return marked - def sync_from_api(self, client: ModelHubClient | ModelHubClientPool) -> int: + def sync_from_api( + self, + client: ModelHubClient | ModelHubClientPool, + *, + task_contexts: dict[str, dict[str, Any]] | None = None, + ) -> int: try: begin = self._last_sync_time end = utc_now() @@ -350,13 +374,32 @@ class OutcomeTracker: if not task_id: continue + task = self._enrich_history_task(task, (task_contexts or {}).get(task_id)) + existing = self._by_task_id.get(task_id) if existing is not None: + metadata_before = ( + existing.get("modelId"), + existing.get("targetGpu"), + existing.get("framework"), + existing.get("taskType"), + existing.get("modelProfile"), + ) + self._merge_record_metadata(existing, task) + metadata_changed = metadata_before != ( + existing.get("modelId"), + existing.get("targetGpu"), + existing.get("framework"), + existing.get("taskType"), + existing.get("modelProfile"), + ) if existing.get("outcome") in {"pending", "policy_cancelled"}: self._update_record_from_task(existing, task) if existing.get("outcome") == "failed" and existing.get("logCosUrl"): enrichment_candidates.append(existing) updated_count += 1 + elif metadata_changed: + updated_count += 1 else: status = str(task.get("status") or "").lower() if status in TERMINAL_TASK_STATUSES: @@ -393,6 +436,7 @@ class OutcomeTracker: if updated_count or enrichment_attempts: self._last_sync_time = end self._rebuild_failed_index() + write_jsonl(self.recent_path, self._recent_records) self.save() return updated_count @@ -630,6 +674,18 @@ class OutcomeTracker: now = now or utc_now() return failed_at >= now - timedelta(hours=max(0, int(cooldown_hours))) + def has_non_transformers_failure(self, model_id: str) -> bool: + """Return platform evidence that unlocks prerequisite-gated transformers.""" + for record in [*self._recent_records, *self._records]: + if str(record.get("modelId") or "") != str(model_id): + continue + if record.get("outcome") != "failed": + continue + framework = str(record.get("framework") or "").strip().casefold() + if framework and framework != "transformers": + return True + return False + def get_stats_report(self) -> dict[str, Any]: now_datetime = utc_now() now = now_datetime.isoformat() diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 153bed2f..5d4ed7fb 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -560,6 +560,13 @@ def run_poll_loop( poll_run_dir = make_run_dir(poll_runs_dir, now) outcome_tracker = outcome_tracker or OutcomeTracker(Path(base_args.outcomes_path)) + recovered_contexts = _load_task_compatibility_contexts( + outcome_tracker, + ledger_path=Path(base_args.ledger_path), + ) + repaired_contexts = outcome_tracker.merge_task_contexts(recovered_contexts) + if repaired_contexts: + log(f"[outcome-recovery] metadata_repaired={repaired_contexts}") OUTCOME_SYNC_INTERVAL = 3 STATS_PRINT_INTERVAL = 10 @@ -652,7 +659,13 @@ def run_poll_loop( and not getattr(base_args, "skip_outcome_sync", False) ): try: - synced = outcome_tracker.sync_from_api(modelhub_client) + synced = outcome_tracker.sync_from_api( + modelhub_client, + task_contexts=_load_task_compatibility_contexts( + outcome_tracker, + ledger_path=Path(base_args.ledger_path), + ), + ) outcome_synced_this_cycle = True if synced > 0: log(f"[poll] cycle={cycles} outcome_sync_updated={synced}") @@ -702,7 +715,13 @@ def run_poll_loop( not outcome_synced_this_cycle and not getattr(base_args, "skip_outcome_sync", False) ): - synced_before_cleanup = outcome_tracker.sync_from_api(modelhub_client) + synced_before_cleanup = outcome_tracker.sync_from_api( + modelhub_client, + task_contexts=_load_task_compatibility_contexts( + outcome_tracker, + ledger_path=Path(base_args.ledger_path), + ), + ) if synced_before_cleanup: log( f"[queue-cleanup] outcome_sync_updated={synced_before_cleanup}" @@ -920,7 +939,13 @@ def run_poll_loop( time.sleep(cooldown) try: - outcome_tracker.sync_from_api(modelhub_client) + outcome_tracker.sync_from_api( + modelhub_client, + task_contexts=_load_task_compatibility_contexts( + outcome_tracker, + ledger_path=Path(base_args.ledger_path), + ), + ) except Exception: pass outcome_tracker.save() diff --git a/modelhub_submmit_api/routing_engine.py b/modelhub_submmit_api/routing_engine.py index ac4ef7fe..7e101456 100644 --- a/modelhub_submmit_api/routing_engine.py +++ b/modelhub_submmit_api/routing_engine.py @@ -138,6 +138,20 @@ class SuccessFirstRoutingEngine: selected: list[dict[str, Any]] = [] for routes in by_model.values(): + # The platform permits some transformers submissions only after a + # non-transformers attempt has failed. Prefer an immediately + # executable route whenever one exists; transformers remains a + # fallback for tasks/models with no other official route. + non_transformers = [ + item for item in routes if str(item.get("framework") or "").casefold() != "transformers" + ] + transformers_unlocked = any( + bool(item.get("transformersPrerequisiteSatisfied")) + for item in routes + if str(item.get("framework") or "").casefold() == "transformers" + ) + if non_transformers and not transformers_unlocked: + routes = non_transformers best = max(float(item["routingSuccessLowerBound"]) for item in routes) close = [item for item in routes if best - float(item["routingSuccessLowerBound"]) <= 0.05] close.sort( diff --git a/modelhub_submmit_api/state_sync.py b/modelhub_submmit_api/state_sync.py index d2545444..3c0ad711 100644 --- a/modelhub_submmit_api/state_sync.py +++ b/modelhub_submmit_api/state_sync.py @@ -26,9 +26,10 @@ STATE_SCHEMA_VERSION = 1 DEFAULT_REMOTE = "https://dev.modelhub.org.cn/CoolBoy/submmit.git" DEFAULT_BRANCH = "agent-state" DEFAULT_ARCHIVE_BRANCH_PREFIX = "agent-archive" -DEFAULT_BATCH_SIZE = 20 +DEFAULT_BATCH_SIZE = 100 DEFAULT_RETENTION_DAYS = 30 -DEFAULT_HISTORY_DEPTH = 20 +DEFAULT_HISTORY_DEPTH = 200 +DEFAULT_RECENT_TERMINAL_INTENTS = 200 # Only these runtime files may cross the trust boundary into the state branch. # Credentials, raw stdout, downloaded archives and run directories are excluded. @@ -154,6 +155,9 @@ class StateGitSync: self._mutex = threading.Lock() self._workspace: Path | None = None self._expected_remote_oid: str | None = None + self._pending_push_oid: str | None = None + self._pending_generation: int | None = None + self._pending_manifest: dict[str, Any] | None = None @property def state_dir(self) -> Path: @@ -216,8 +220,8 @@ class StateGitSync: return oid.decode("ascii") if oid else None def _sync_archive_pending(self) -> None: - pending_root = self.state_dir / "archive_pending" / "outcomes" - files = sorted(path for path in pending_root.glob("*/*.jsonl.gz") if path.is_file()) + pending_root = self.state_dir / "archive_pending" + files = sorted(path for path in pending_root.glob("*/*/*.jsonl.gz") if path.is_file()) if not files: return by_month: dict[str, list[Path]] = {} @@ -251,10 +255,11 @@ class StateGitSync: manifest = {} archived = manifest.get("files") if isinstance(manifest.get("files"), dict) else {} for source in month_files: - destination = workspace / "outcomes" / source.name + archive_kind = source.parent.parent.name + destination = workspace / archive_kind / source.name destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, destination) - archived[f"outcomes/{source.name}"] = { + archived[f"{archive_kind}/{source.name}"] = { "sha256": _sha256_file(destination), "bytes": destination.stat().st_size, } @@ -273,7 +278,7 @@ class StateGitSync: if any(status.staged.get(kind) for kind in ("add", "delete", "modify")): porcelain.commit( repo, - message=f"archive: outcomes {month}".encode("utf-8"), + message=f"archive: durable records {month}".encode("utf-8"), author=self._author, committer=self._author, ) @@ -404,6 +409,49 @@ class StateGitSync: self._expected_remote_oid = None return self.restore() + def _archive_intents(self, records: list[dict[str, Any]]) -> str | None: + if not records: + return None + now = _utc_now() + month = now.strftime("%Y-%m") + name = f"{now.strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:10]}.jsonl.gz" + path = self.state_dir / "archive_pending" / "attempts" / month / name + path.parent.mkdir(parents=True, exist_ok=True) + import gzip + + with gzip.open(path, "wt", encoding="utf-8", compresslevel=6) as handle: + for record in records: + safe = { + key: value + for key, value in record.items() + if not any(marker in key.casefold() for marker in ("token", "password", "authorization")) + } + handle.write(json.dumps(safe, ensure_ascii=False, sort_keys=True) + "\n") + return f"{month}/{name}" + + def _compact_intents(self) -> int: + records = read_jsonl(self.intents_path) + if len(records) <= DEFAULT_RECENT_TERMINAL_INTENTS: + return 0 + active_statuses = {"pending", "submitted", "recovered_active"} + active = [row for row in records if str(row.get("status") or "") in active_statuses] + terminal = [row for row in records if str(row.get("status") or "") not in active_statuses] + terminal.sort( + key=lambda row: str(row.get("completedAt") or row.get("createdAt") or ""), + reverse=True, + ) + retained_terminal = terminal[:DEFAULT_RECENT_TERMINAL_INTENTS] + archived = terminal[DEFAULT_RECENT_TERMINAL_INTENTS:] + if not archived: + return 0 + self._archive_intents(archived) + write_jsonl(self.intents_path, [*active, *retained_terminal]) + self.log( + f"[state-compact] intents_archived={len(archived)} " + f"active={len(active)} recent_terminal={len(retained_terminal)}" + ) + return len(archived) + def _event_files(self) -> list[Path]: event_dir = self.state_dir / "events" if not event_dir.exists(): @@ -588,6 +636,7 @@ class StateGitSync: unresolved += 1 retention_cutoff = now - timedelta(days=self.retention_days) retained: list[dict[str, Any]] = [] + expired: list[dict[str, Any]] = [] for intent in intents: completed_text = intent.get("completedAt") if not completed_text: @@ -602,12 +651,16 @@ class StateGitSync: completed_at = completed_at.replace(tzinfo=timezone.utc) if completed_at >= retention_cutoff: retained.append(intent) + else: + expired.append(intent) + self._archive_intents(expired) write_jsonl(self.intents_path, retained) self.record_active_tasks(enriched) return {"active": len(enriched), "reconciled": reconciled, "unresolved": unresolved} def _copy_snapshot(self) -> dict[str, str]: assert self._workspace is not None + self._compact_intents() checksums: dict[str, str] = {} for relative in STATE_ALLOWLIST: source = self.project_root / relative @@ -634,7 +687,21 @@ class StateGitSync: if isinstance(sanitized_market, dict): # A restored snapshot must refresh official configs before routing. sanitized_market["frameworkUpdatedAt"] = None + # Raw community rows are a disposable refresh cache. The + # aggregated framework/GPU statistics are the durable input. + sanitized_market["communitySample"] = {} write_json(destination, sanitized_market) + elif relative == ".modelhub_state/official_capabilities.json": + payload = read_json(source) + cache = payload.get("modelGpuTaskTypes") if isinstance(payload, dict) else None + if isinstance(cache, dict) and len(cache) > 1500: + ordered = sorted( + cache.items(), + key=lambda pair: str((pair[1] or {}).get("updatedAt") or ""), + reverse=True, + ) + payload["modelGpuTaskTypes"] = dict(ordered[:1500]) + write_json(destination, payload) elif relative in {"outcomes/submissions.jsonl", ".modelhub_state/recent_outcomes.jsonl"}: sanitized_outcomes: list[dict[str, Any]] = [] for row in read_jsonl(source): @@ -666,16 +733,69 @@ class StateGitSync: path.unlink(missing_ok=True) return checksums + def _push_pending(self, *, phase: str) -> bool: + if self._pending_push_oid is None: + return True + assert self._workspace is not None + repo = Repo(str(self._workspace)) + remote_oid = self._remote_oid() + if remote_oid == self._pending_push_oid: + pushed = True + elif remote_oid != self._expected_remote_oid: + raise StateSyncError("state branch changed remotely; refusing to overwrite a newer snapshot") + else: + porcelain.push( + repo, + self.remote, + refspecs=f"HEAD:refs/heads/{self.branch}", + force=True, + outstream=io.BytesIO(), + errstream=io.BytesIO(), + **self._auth_kwargs(), + ) + pushed = self._remote_oid() == self._pending_push_oid + if not pushed: + raise StateSyncError("state branch verification failed after push") + self._expected_remote_oid = self._pending_push_oid + self.generation = int(self._pending_generation or self.generation) + manifest = self._pending_manifest or {} + self.last_sync_at = manifest.get("updatedAt") + self._pending_push_oid = None + self._pending_generation = None + self._pending_manifest = None + self.last_error = None + self.healthy = True + self.log(f"[state-sync] generation={self.generation} phase={phase} status=ok") + self._sync_archive_pending_safely() + return True + def sync(self, phase: str) -> bool: with self._mutex: try: if self._workspace is None: raise StateSyncError("state workspace is not initialized") + # A failed network push is retried byte-for-byte. Do not create + # another commit or generation for the same state transition. + if self._pending_push_oid is not None: + return self._push_pending(phase=phase) + previous_manifest: dict[str, Any] = {} + try: + previous_manifest = read_json(self._workspace / "manifest.json") + except (FileNotFoundError, ValueError, TypeError): + pass checksums = self._copy_snapshot() - self.generation += 1 + if ( + previous_manifest.get("checksums") == checksums + and previous_manifest.get("agentVersion") == AGENT_VERSION + ): + self.healthy = True + self.last_error = None + self._sync_archive_pending_safely() + return True + next_generation = self.generation + 1 manifest = { "schemaVersion": STATE_SCHEMA_VERSION, - "generation": self.generation, + "generation": next_generation, "updatedAt": _utc_now().isoformat(), "agentVersion": AGENT_VERSION, "writerId": self.writer_id, @@ -696,7 +816,7 @@ class StateGitSync: return True porcelain.commit( repo, - message=f"state: generation {self.generation} ({phase})".encode("utf-8"), + message=f"state: generation {next_generation} ({phase})".encode("utf-8"), author=self._author, committer=self._author, ) @@ -707,32 +827,15 @@ class StateGitSync: porcelain.add(repo) porcelain.commit( repo, - message=f"state: compacted generation {self.generation}".encode("utf-8"), + message=f"state: compacted generation {next_generation}".encode("utf-8"), author=self._author, committer=self._author, ) - current_remote_oid = self._remote_oid() - if current_remote_oid != self._expected_remote_oid: - raise StateSyncError("state branch changed remotely; refusing to overwrite a newer snapshot") - porcelain.push( - repo, - self.remote, - refspecs=f"HEAD:refs/heads/{self.branch}", - force=True, - outstream=io.BytesIO(), - errstream=io.BytesIO(), - **self._auth_kwargs(), - ) local_oid = repo.head().decode("ascii") - if self._remote_oid() != local_oid: - raise StateSyncError("state branch verification failed after push") - self._expected_remote_oid = local_oid - self.last_sync_at = manifest["updatedAt"] - self.last_error = None - self.healthy = True - self.log(f"[state-sync] generation={self.generation} phase={phase} status=ok") - self._sync_archive_pending_safely() - return True + self._pending_push_oid = local_oid + self._pending_generation = next_generation + self._pending_manifest = manifest + return self._push_pending(phase=phase) except Exception as exc: self.healthy = False self.last_error = _safe_text(exc) diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 6219aa10..452e3086 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.22.1" +AGENT_VERSION = "2026.09.04.1" diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 041bd4a2..98254ce3 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -319,6 +319,45 @@ class ClientPoolConcurrencyTests(unittest.TestCase): self.assertEqual("age_policy_skipped", result["reason"]) self.assertEqual([], client.submitted) + def test_submit_capacity_exhaustion_is_deferred_not_failed(self) -> None: + class CapacityClient: + @staticmethod + def add_task(_payload): # noqa: ANN001, ANN205 + raise ModelHubAPIError("当前等待中或运行中的异步模型验证任务数量已达上限（100）") + + result = submit_candidate( + { + "repoId": "owner/model", + "modelAddress": "https://modelscope.cn/models/owner/model", + "targetGpu": "gpu-a", + "framework": "vllm", + "taskType": "text-generation", + "configParams": "safe", + }, + CapacityClient(), # type: ignore[arg-type] + ) + self.assertEqual("capacity_deferred", result["outcome"]) + self.assertEqual("account_capacity_saturated", result["reason"]) + + def test_transformers_platform_prerequisite_is_deferred_not_failed(self) -> None: + class PrerequisiteClient: + @staticmethod + def add_task(_payload): # noqa: ANN001, ANN205 + raise ModelHubAPIError("该模型必须在非transformers框架验证失败后才可以开启transformers框架验证任务") + + result = submit_candidate( + { + "repoId": "owner/model", + "modelAddress": "https://modelscope.cn/models/owner/model", + "targetGpu": "gpu-a", + "framework": "transformers", + "taskType": "text-generation", + "configParams": "safe", + }, + PrerequisiteClient(), # type: ignore[arg-type] + ) + self.assertEqual("framework_prerequisite_deferred", result["outcome"]) + def test_age_policy_deferred_candidate_is_skipped_not_failed(self) -> None: class AdmissionRaceClient: @staticmethod diff --git a/tests/test_super_agent.py b/tests/test_super_agent.py index 4b3d6a7a..d0386c32 100644 --- a/tests/test_super_agent.py +++ b/tests/test_super_agent.py @@ -193,6 +193,63 @@ class SuperAgentTests(unittest.TestCase): self.assertEqual("reliable", ordered[0]["targetGpu"]) self.assertGreater(ordered[0]["routingSuccessBand"], ordered[1]["routingSuccessBand"]) + def test_transformers_is_fallback_when_an_executable_framework_exists(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + engine = SuccessFirstRoutingEngine(Path(temporary_dir) / "routing.json", log_fn=lambda _: None) + candidates = [ + { + "repoId": "owner/model", + "targetGpu": "gpu-a", + "framework": "transformers", + "taskType": "text-generation", + "frameworkMarketSamples": 1000, + "frameworkMarketSuccessRate": 0.95, + }, + { + "repoId": "owner/model", + "targetGpu": "gpu-b", + "framework": "vllm", + "taskType": "text-generation", + "frameworkMarketSamples": 1000, + "frameworkMarketSuccessRate": 0.40, + }, + ] + ordered = engine.order_candidates(candidates) + self.assertEqual(1, len(ordered)) + self.assertEqual("vllm", ordered[0]["framework"]) + + candidates[0]["transformersPrerequisiteSatisfied"] = True + unlocked = engine.order_candidates(candidates) + self.assertEqual("transformers", unlocked[0]["framework"]) + + def test_outcome_context_repairs_framework_missing_from_history_api(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + tracker = OutcomeTracker(root / "outcomes.jsonl") + tracker._records = [ + { + "taskId": "42", + "modelId": "owner/model", + "targetGpu": "gpu-a", + "framework": "", + "taskType": "", + "outcome": "failed", + } + ] + tracker._rebuild_indexes() + repaired = tracker.merge_task_contexts( + { + "42": { + "framework": "vllm", + "taskType": "text-generation", + "modelProfile": {"modelType": "Qwen2ForCausalLM"}, + } + } + ) + self.assertEqual(1, repaired) + self.assertEqual("vllm", tracker._by_task_id["42"]["framework"]) + self.assertEqual("Qwen2ForCausalLM", tracker._by_task_id["42"]["modelProfile"]["modelType"]) + def test_modelscope_metadata_and_model_card_lineage_are_structured(self) -> None: item = { "id": "owner/model", @@ -270,6 +327,11 @@ class SuperAgentTests(unittest.TestCase): ] ) self.assertIsNotNone(batch_id) + generation = manager.generation + remote_head = manager._remote_oid() + self.assertTrue(manager.sync("unchanged_cycle")) + self.assertEqual(generation, manager.generation) + self.assertEqual(remote_head, manager._remote_oid()) self.assertFalse(pending_archive.exists()) archive_refs = porcelain.ls_remote(str(remote)).refs self.assertIn(b"refs/heads/agent-archive-2026-08", archive_refs) @@ -296,6 +358,63 @@ class SuperAgentTests(unittest.TestCase): self.assertNotIn("must-not-be-copied", state_text) restored.close() + def test_failed_state_push_retries_the_same_commit_and_generation(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + remote = root / "remote.git" + project = root / "project" + project.mkdir() + porcelain.init(remote, bare=True) + manager = StateGitSync( + project_root=project, + credentials={"username": "u", "email": "e@example.com", "password": "p"}, + remote=str(remote), + log_fn=lambda _: None, + ) + manager.acquire_process_lock() + self.assertTrue(manager.restore()) + write_json(project / ".modelhub_state" / "account_capacity.json", {"version": 1}) + with patch("state_sync.porcelain.push", side_effect=RuntimeError("temporary failure")): + self.assertFalse(manager.sync("cycle")) + pending_oid = manager._pending_push_oid + self.assertIsNotNone(pending_oid) + self.assertEqual(0, manager.generation) + self.assertTrue(manager.sync("retry")) + self.assertEqual(1, manager.generation) + self.assertEqual(pending_oid, manager._remote_oid()) + manager.close() + + def test_terminal_intents_are_bounded_and_archived(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + intent_path = root / ".modelhub_state" / "recovery_intents.jsonl" + write_jsonl( + intent_path, + [ + { + "intentId": str(index), + "status": "failed", + "createdAt": f"2026-08-01T00:{index % 60:02d}:00+00:00", + "completedAt": f"2026-08-02T00:{index % 60:02d}:00+00:00", + } + for index in range(250) + ] + + [{"intentId": "pending", "status": "pending"}], + ) + manager = StateGitSync( + project_root=root, + credentials={"username": "u", "email": "e@example.com", "password": "p"}, + remote="unused", + log_fn=lambda _: None, + ) + self.assertEqual(50, manager._compact_intents()) + retained = read_jsonl(intent_path) + self.assertEqual(201, len(retained)) + self.assertEqual(1, sum(row.get("status") == "pending" for row in retained)) + shard = next((root / ".modelhub_state" / "archive_pending" / "attempts").rglob("*.jsonl.gz")) + with gzip.open(shard, "rt", encoding="utf-8") as handle: + self.assertEqual(50, len(handle.readlines())) + def test_failed_intent_push_returns_no_batch_id(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: manager = StateGitSync(