From e38d5cf82352cf65ef4e4962305290a3b624fb49 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Mon, 26 Oct 2026 14:56:55 +0800 Subject: [PATCH] fix: bound unproven profile submissions and support single-worker takeover --- README.md | 45 ++++++++ main.py | 36 +++++++ modelhub_submmit_api/candidate_preflight.py | 6 ++ modelhub_submmit_api/history_stats.py | 2 +- modelhub_submmit_api/main.py | 11 ++ modelhub_submmit_api/models.py | 23 ++++ modelhub_submmit_api/outcome_tracker.py | 36 ++++++- modelhub_submmit_api/poll_runner.py | 9 ++ modelhub_submmit_api/profile_admission.py | 41 +++++++ modelhub_submmit_api/routing_engine.py | 2 +- modelhub_submmit_api/state_sync.py | 62 +++++++++-- modelhub_submmit_api/version.py | 2 +- tests/test_takeover_accuracy.py | 114 ++++++++++++++++++++ 13 files changed, 376 insertions(+), 13 deletions(-) create mode 100644 modelhub_submmit_api/profile_admission.py create mode 100644 tests/test_takeover_accuracy.py diff --git a/README.md b/README.md index b13dbb9e9c..934eb03e2a 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,51 @@ This repository is packaged for the ModelHub XC agent platform. +## Single-worker takeover release 2026.10.26.1 + +This release is intended to replace the competing workers, using all configured +accounts in one process. Deployment must stop the old **agent process**, not +its already-submitted validation tasks. Code cannot reserve capacity while an +old worker continues to refill the same accounts. No live stop or cancellation +is performed by installing this release. + +Keep `MODELHUB_STATE_SYNC_BRANCH=agent-state-accuracy-v2` to resume the newest +checkpoint. Startup adopts the live waiting/running queue by task ID, without +resubmission or attributing old work to the new version. It retains existing +aggregate evidence; do not add the old branch's lifetime counters to it, which +would double-count shared tasks. Historical tasks without encoding/provenance +remain unknown. Queue cancellation remains disabled by default during takeover; +existing tasks continue occupying capacity until the platform finishes them. + +Changes: + +- Recognize MLX metadata and top-level quantization configuration. Unknown packed + formats are deferred before submission to conventional backends. GGUF routes + retain their existing file-specific checks. No repository-name heuristics. +- Persist an encoding signature so FP8 and INT4 variants of compressed-tensors + do not share a feedback-admission budget. Until the exact architecture, + encoding, GPU and framework has three successes and at least 50% observed + non-platform success, admit at most two pending tasks per group, including + tasks selected in the same concurrent batch. Deferrals are not failures or + permanent exclusions; completed tasks release the budget. This cap only + constrains already-eligible candidates; it does not create exploratory routes. +- In the same success-confidence band, completion-speed estimates precede + community popularity when ordering candidates. Quality gates remain enabled. +- Persist per-version terminal results and print `[feedback]` with success, + failure and pending counts. Unexplained failed statuses no longer pretend to + identify a parameter/template defect. +- Close Git repository handles, throttle idle-cycle snapshots to once per + minute (intent/result writes stay synchronous), bound history traversal, and + rotate the disposable local clone after verified history compaction so old + unreachable objects do not accumulate indefinitely. +- Log worker RSS/high-water memory and available cgroup memory counters once + per minute; retain the last sample and exit-time OOM counters with crashes. + +The thresholds above are conservative policy settings, not a guarantee of a +particular success rate. Evaluate actual terminal outcomes separately from +waiting tasks after deployment. A full inherited queue will not immediately +produce new capacity just because the old agent process has stopped. + ## Accuracy release 2026.10.18.1 This release rejects malformed YAML command arrays, including the observed diff --git a/main.py b/main.py index 86d1795651..1d45043e4f 100644 --- a/main.py +++ b/main.py @@ -187,6 +187,33 @@ def _write_supervisor_readiness(reason: str, *, exit_code: int | None = None) -> os.replace(temporary, READINESS_PATH) +_worker_memory_sample: dict[str, object] = {} +_worker_memory_sample_at = 0.0 + + +def _sample_worker_memory(now: float) -> None: + global _worker_memory_sample, _worker_memory_sample_at + if now - _worker_memory_sample_at < 60: + return + _worker_memory_sample_at = now + sample: dict[str, object] = {"sampledAt": datetime.now(timezone.utc).isoformat()} + try: + pid = int(getattr(worker, "pid", 0)) + for line in Path(f"/proc/{pid}/status").read_text().splitlines(): + if line.startswith(("VmRSS:", "VmHWM:")): + name, value, *_ = line.split() + sample[name.rstrip(":") + "KiB"] = int(value) + except (OSError, ValueError, TypeError): + pass + for name in ("memory.current", "memory.peak", "memory.max", "memory.events"): + try: + sample[name] = (Path("/sys/fs/cgroup") / name).read_text().strip()[:500] + except OSError: + pass + _worker_memory_sample = sample + print("[worker-memory] " + json.dumps(sample, separators=(",", ":")), flush=True) + + def _record_worker_crash(*, exit_code: int | None, uptime_seconds: float, reason: str) -> None: WORKER_CRASH_PATH.parent.mkdir(parents=True, exist_ok=True) event = { @@ -195,13 +222,21 @@ def _record_worker_crash(*, exit_code: int | None, uptime_seconds: float, reason "uptimeSeconds": round(max(0.0, uptime_seconds), 3), "restartCount": worker_restart_count, "reason": reason[:500], + "lastMemorySample": dict(_worker_memory_sample), } + try: + event["cgroupMemoryEventsAtExit"] = Path("/sys/fs/cgroup/memory.events").read_text().strip()[:500] + except OSError: + pass with WORKER_CRASH_PATH.open("a", encoding="utf-8") as handle: handle.write(json.dumps(event, ensure_ascii=False, separators=(",", ":")) + "\n") def _launch_worker(*, now: float | None = None) -> bool: global worker, worker_started_at, worker_last_exit_code + global _worker_memory_sample, _worker_memory_sample_at + _worker_memory_sample = {} + _worker_memory_sample_at = 0.0 now = time.monotonic() if now is None else now cmd = _worker_command() print("starting submission worker: " + " ".join(cmd), flush=True) @@ -227,6 +262,7 @@ def _supervise_worker(*, now: float | None = None) -> None: if worker is not None: return_code = worker.poll() if return_code is None: + _sample_worker_memory(now) if worker_started_at is not None and now - worker_started_at >= WORKER_STABLE_SECONDS: worker_restart_count = 0 return diff --git a/modelhub_submmit_api/candidate_preflight.py b/modelhub_submmit_api/candidate_preflight.py index d7c2e42da6..2ba9330ddb 100644 --- a/modelhub_submmit_api/candidate_preflight.py +++ b/modelhub_submmit_api/candidate_preflight.py @@ -189,6 +189,7 @@ class CandidatePreflightAdvisor: "modelType": inspection.model_type, "architectures": inspection.architectures, "quantizationMethod": inspection.quantization_method, + "quantizationSignature": inspection.quantization_signature, "estimatedLoadBytes": inspection.estimated_load_bytes(framework), "repositoryOnDiskBytes": inspection.repository_size_bytes, "recursiveRepositoryBytes": inspection.exact_repository_size_bytes, @@ -199,6 +200,11 @@ class CandidatePreflightAdvisor: with self._lock: self._assessed += 1 + if framework != "llamacpp" and inspection.quantization_method in {"mlx", "unknown"}: + return self._hard_block( + config_params, "preflight_unverified_weight_format", warnings, ambiguous, metadata, + ) + learned_block = self._matching_architecture_block( inspection=inspection, target_gpu=target_gpu, diff --git a/modelhub_submmit_api/history_stats.py b/modelhub_submmit_api/history_stats.py index a3d1993eb4..e5c7a2f3ee 100644 --- a/modelhub_submmit_api/history_stats.py +++ b/modelhub_submmit_api/history_stats.py @@ -257,7 +257,7 @@ def classify_failure(record: dict[str, Any]) -> str: status = str(record.get("status") or "").lower() verify_result = _numeric_verify_result(record.get("verifyResult")) if status in {"failed", "error", "cancelled", "canceled", "rejected", "timeout"}: - return "参数/模板问题" + return "任务失败_原因未确认" if status in WAITING_STATUSES or status in RUNNING_STATUSES or verify_result is None: return "排队中/未知" if verify_result is not None and verify_result < 0: diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 00dac95eea..c2347cebb5 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -47,6 +47,7 @@ from official_capabilities import ( OfficialCapabilityUnavailable, ) from outcome_tracker import DEFAULT_OUTCOMES_PATH, OutcomeTracker +from profile_admission import filter_profile_batch from routing_engine import DEFAULT_ROUTING_STATE_PATH, SuccessFirstRoutingEngine from submission_claims import DEFAULT_CLAIMS_PATH, SubmissionClaimStore, candidate_key, diversify_candidates from submission_exclusions import DEFAULT_SUBMISSION_EXCLUSIONS_PATH, SubmissionExclusionStore @@ -737,6 +738,8 @@ def process_model_for_candidates( inspection = hf_discovery.inspect_model(model) except Exception as exc: return [], skipped, [{"repoId": model.repo_id, "reason": str(exc)}] + if "library:mlx" in {str(tag).lower() for tag in model.tags}: + inspection.model_card_metadata = {**inspection.model_card_metadata, "library_name": "mlx"} def evaluate_pending_routes(pending_routes: list[tuple[str, list[str], set[str]]]) -> int: before_count = len(candidates) @@ -1703,6 +1706,7 @@ def run_submission( owner_id=instance_id, ) attempted_keys: set[str] = set() + profile_stats = outcome_tracker.get_stats_report() attempt_multiplier = max(1, int(getattr(args, "scan_multiplier", 4) or 1)) max_submit_attempts = min( len(diversified_candidates), @@ -1721,6 +1725,13 @@ def run_submission( and not submission_exclusion_store.is_blocked(candidate["repoId"], candidate["targetGpu"]) ] remaining_candidates = one_candidate_per_model(remaining_candidates) + remaining_candidates, profile_deferred = filter_profile_batch( + remaining_candidates, outcome_tracker.pending_records(), profile_stats, + ) + for candidate in profile_deferred: + attempted_keys.add(candidate_key(candidate)) + skipped.append({"repoId": candidate["repoId"], "targetGpu": candidate["targetGpu"], + "reason": "profile_feedback_deferred", "deferred": True}) desired_count = min( target_submit_count - len(submitted), max_submit_attempts - len(attempted_candidates), diff --git a/modelhub_submmit_api/models.py b/modelhub_submmit_api/models.py index f62bae3312..b867f1b488 100644 --- a/modelhub_submmit_api/models.py +++ b/modelhub_submmit_api/models.py @@ -4,6 +4,8 @@ from dataclasses import dataclass, field from datetime import datetime from pathlib import PurePosixPath from typing import Any +import hashlib +import json @dataclass(frozen=True) @@ -114,13 +116,34 @@ class ModelInspection: @property def quantization_method(self) -> str | None: + if str(self.model_card_metadata.get("library_name") or "").lower() == "mlx": + return "mlx" value = self.model_config.get("quantization_config") if isinstance(value, dict): method = value.get("quant_method") or value.get("quantization_method") if method not in (None, ""): return str(method).strip().lower() + if value: + return "unknown" + # MLX commonly uses a top-level quantization object, without + # quant_method. Never treat packed weights as ordinary FP weights. + value = self.model_config.get("quantization") + if isinstance(value, dict) and value: + if "bits" in value and "group_size" in value: + return "mlx" + return "unknown" return None + @property + def quantization_signature(self) -> str: + value = self.model_config.get("quantization_config") or self.model_config.get("quantization") or {} + if not isinstance(value, dict): + value = {} + # compressed-tensors covers several incompatible formats. Preserve + # the encoding configuration, not just its broad method name. + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return f"{self.quantization_method or 'none'}:{hashlib.sha256(encoded.encode()).hexdigest()[:16]}" + @property def max_context_length(self) -> int | None: for key in ( diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index 7ce4e65ee7..3243e8a735 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -20,7 +20,7 @@ from common import append_jsonl, parse_datetime, read_json, read_jsonl, update_j from failure_log_inspector import fetch_and_classify_failure_log from history_stats import classify_failure, is_failure, is_success from llm_classifier import LLMAssistedClassifier -from modelhub_client import ModelHubClient, ModelHubClientPool +from modelhub_client import ModelHubClient, ModelHubClientPool, is_active_task from task_registry import task_type_from_history_task from common import runtime_instance_id from version import AGENT_VERSION @@ -361,6 +361,31 @@ class OutcomeTracker: pending_gpus.add(target_gpu) return len(pending_gpus) + def pending_records(self) -> list[dict[str, Any]]: + return [row for row in self._records if row.get("outcome") == "pending"] + + def adopt_active_tasks(self, tasks, contexts=None) -> int: + """Adopt by task ID without inventing ownership or resubmitting work.""" + adopted = 0 + for task in tasks: + task_id = str(task.get("taskId") or "") + if not task_id or task_id in self._by_task_id or not is_active_task(task): + continue + enriched = self._enrich_history_task(task, (contexts or {}).get(task_id)) + record = {"taskId": task_id, "modelId": enriched.get("modelId"), + "targetGpu": enriched.get("gpuType"), "framework": enriched.get("framework"), + "taskType": enriched.get("taskType") or task_type_from_history_task(enriched), + "modelProfile": enriched.get("modelProfile") or {}, + "status": enriched.get("status"), "outcome": "pending", + "submitTime": enriched.get("createTime") or enriched.get("submitTime"), + "lastSyncTime": _now_iso()} + self._records.append(record) + self._by_task_id[task_id] = record + self._by_model_gpu[(record["modelId"], record["targetGpu"])].append(record) + append_jsonl(self.path, record) + adopted += 1 + return adopted + def mark_policy_cancellations(self, decisions: list[dict[str, Any]]) -> int: """Persist intentional task stops so they never become failure evidence.""" marked = 0 @@ -760,11 +785,14 @@ class OutcomeTracker: framework_groups: dict[str, list[dict[str, Any]]] = defaultdict(list) combo_groups: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) profile_groups: dict[tuple[str, str, str, str, str], list[dict[str, Any]]] = defaultdict(list) + encoding_groups: dict[str, list[dict[str, Any]]] = defaultdict(list) + agent_groups: dict[str, list[dict[str, Any]]] = defaultdict(list) sized_profile_groups: dict[tuple[str, str, str, str, str, int], list[dict[str, Any]]] = defaultdict(list) recent_combo_groups: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list) recent_profile_groups: dict[tuple[str, str, str, str, str], list[dict[str, Any]]] = defaultdict(list) for record in terminal: + agent_groups[str(record.get("agentVersion") or "unknown")].append(record) gpu = record.get("targetGpu") or "unknown" fw = record.get("framework") or "unknown" tt = record.get("taskType") or "unknown" @@ -776,6 +804,8 @@ class OutcomeTracker: if model_type: quantization = str(profile.get("quantizationMethod") or "none").strip() profile_groups[(gpu, fw, tt, model_type, quantization)].append(record) + if profile.get("quantizationSignature"): + encoding_groups[f"{gpu}|{fw}|{tt}|{model_type}|{profile['quantizationSignature']}"].append(record) try: load_bytes = max(1, int(profile.get("estimatedLoadBytes") or 0)) except (TypeError, ValueError): @@ -914,6 +944,8 @@ class OutcomeTracker: "combinationStats": combination_stats, "recentCombinationStats": recent_combination_stats, "profileCombinationStats": profile_combination_stats, + "encodingCombinationStats": {key: _summarize(rows) for key, rows in encoding_groups.items()}, + "agentSummaries": {key: _summarize(rows) for key, rows in agent_groups.items()}, "recentProfileCombinationStats": recent_profile_combination_stats, "sizedProfileCombinationStats": sized_profile_combination_stats, "architectureCompatibilityBlocks": architecture_blocks, @@ -944,6 +976,8 @@ class OutcomeTracker: "frameworkSummaries", "combinationStats", "profileCombinationStats", + "encodingCombinationStats", + "agentSummaries", "sizedProfileCombinationStats", ): base_items = baseline.get(field) if isinstance(baseline.get(field), dict) else {} diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 41d6e44084..663a18ab4e 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -668,6 +668,7 @@ def run_poll_loop( modelhub_client = modelhub_client or _build_modelhub_client(base_args) template_selector = template_selector or TemplateSelector() + active_tasks = [] if state_sync is not None and state_sync.healthy: try: active_tasks = ( @@ -698,6 +699,8 @@ def run_poll_loop( ledger_path=Path(base_args.ledger_path), ) repaired_contexts = outcome_tracker.merge_task_contexts(recovered_contexts) + adopted = outcome_tracker.adopt_active_tasks(active_tasks, recovered_contexts) + log(f"[takeover] active_adopted={adopted} existing_queue=preserved ownership=unchanged") if repaired_contexts: log(f"[outcome-recovery] metadata_repaired={repaired_contexts}") OUTCOME_SYNC_INTERVAL = 3 @@ -889,6 +892,12 @@ def run_poll_loop( if synced > 0: log(f"[poll] cycle={cycles} outcome_sync_updated={synced}") sync_feedback = outcome_tracker.get_stats_report() + own = (sync_feedback.get("agentSummaries") or {}).get(AGENT_VERSION) or {} + own_pending = sum(row.get("agentVersion") == AGENT_VERSION + for row in outcome_tracker.pending_records()) + log(f"[feedback] version={AGENT_VERSION} success={own.get('successCount', 0)} " + f"failed={own.get('failureCount', 0)} pending={own_pending} " + f"unknown_failures={own.get('unresolvedFailureCount', 0)}") active_block_keys = _persist_architecture_blacklist( sync_feedback, path=Path( diff --git a/modelhub_submmit_api/profile_admission.py b/modelhub_submmit_api/profile_admission.py new file mode 100644 index 0000000000..e041fa709b --- /dev/null +++ b/modelhub_submmit_api/profile_admission.py @@ -0,0 +1,41 @@ +"""Bound unproven sibling routes while awaiting actual validation results.""" +from collections import Counter +from typing import Any + + +def profile_key(row: dict[str, Any]) -> tuple[str, ...] | None: + profile = row.get("preflightMetadata") or row.get("modelProfile") or {} + model_type = str(profile.get("modelType") or "").strip() + if not model_type: + return None + return (str(row.get("targetGpu") or row.get("gpuType") or ""), + str(row.get("framework") or ""), str(row.get("taskType") or ""), + model_type, str(profile.get("quantizationSignature") or profile.get("quantizationMethod") or "none")) + + +def filter_profile_batch(candidates, pending, stats, *, limit=2): + """Reservations include the whole proposed batch; deferred rows are retryable. + + Broad public framework statistics are not proof for a specific encoding. + Exact-encoding successes can lift the cap, provided the observed success + rate remains at least 50%. Missing legacy signatures cannot grant this. + """ + counts = Counter(profile_key(row) for row in pending if row.get("outcome") == "pending") + selected, deferred = [], [] + for candidate in candidates: + key = profile_key(candidate) + evidence = (stats.get("encodingCombinationStats") or {}).get("|".join(key or ())) or {} + successes = int(evidence.get("successCount") or 0) + failures = max(0, int(evidence.get("failureCount") or 0) - int(evidence.get("platformFailureCount") or 0)) + proven = successes >= 3 and successes / max(1, successes + failures) >= 0.5 + # Count legacy records with the same method as well, conservatively. + legacy = 0 + if key and ":" in key[-1]: + legacy = counts.get((*key[:-1], key[-1].split(":", 1)[0]), 0) + if key and not proven and counts[key] + legacy >= limit: + deferred.append(candidate) + continue + selected.append(candidate) + if key: + counts[key] += 1 + return selected, deferred diff --git a/modelhub_submmit_api/routing_engine.py b/modelhub_submmit_api/routing_engine.py index 52130d56a8..0b26a74a06 100644 --- a/modelhub_submmit_api/routing_engine.py +++ b/modelhub_submmit_api/routing_engine.py @@ -189,8 +189,8 @@ class SuccessFirstRoutingEngine: selected.sort( key=lambda item: ( -int(item["routingSuccessBand"]), - -int(item.get("routingCommunityModelSuccesses") or 0), -float(item["routingExpectedSuccessPerHour"]), + -int(item.get("routingCommunityModelSuccesses") or 0), -int(item["routingSamples"]), -int(item.get("downloads") or 0), str(item.get("repoId") or ""), diff --git a/modelhub_submmit_api/state_sync.py b/modelhub_submmit_api/state_sync.py index 4d5c51a6a1..bd2dabb23b 100644 --- a/modelhub_submmit_api/state_sync.py +++ b/modelhub_submmit_api/state_sync.py @@ -8,6 +8,9 @@ import os import shutil import tempfile import threading +import time +from contextlib import ExitStack +from itertools import islice import uuid import re from datetime import datetime, timedelta, timezone @@ -171,6 +174,8 @@ class StateGitSync: self._pending_push_oid: str | None = None self._pending_generation: int | None = None self._pending_manifest: dict[str, Any] | None = None + self._last_success_monotonic = 0.0 + self._refresh_workspace_pending = False @property def state_dir(self) -> Path: @@ -236,7 +241,7 @@ class StateGitSync: parent = Path(tempfile.mkdtemp(prefix="modelhub-agent-state-")) workspace = parent / "state" if remote_oid: - porcelain.clone( + cloned = porcelain.clone( self.remote, workspace, branch=self.branch, @@ -245,10 +250,11 @@ class StateGitSync: errstream=io.BytesIO(), **self._auth_kwargs(), ) + cloned.close() else: workspace.mkdir(parents=True) - repo = porcelain.init(workspace) - repo.refs.set_symbolic_ref(b"HEAD", f"refs/heads/{self.branch}".encode("utf-8")) + with porcelain.init(workspace) as repo: + repo.refs.set_symbolic_ref(b"HEAD", f"refs/heads/{self.branch}".encode("utf-8")) self._workspace = workspace self._expected_remote_oid = remote_oid @@ -268,7 +274,7 @@ class StateGitSync: return manifest def restore(self) -> bool: - with self._mutex: + with self._mutex, ExitStack() as resources: try: remote_oid = self._remote_oid() self._create_workspace(remote_oid) @@ -277,7 +283,7 @@ class StateGitSync: manifest = self._validate_manifest(self._workspace) except Exception as current_error: manifest = None - repo = Repo(str(self._workspace)) + repo = resources.enter_context(Repo(str(self._workspace))) commits = [entry.commit.id for entry in repo.get_walker(max_entries=self.history_depth)][1:] for commit in commits: porcelain.reset(repo, "hard", commit) @@ -627,7 +633,10 @@ class StateGitSync: if self._pending_push_oid is None: return True assert self._workspace is not None - repo = Repo(str(self._workspace)) + with Repo(str(self._workspace)) as repo: + return self._push_pending_repo(repo, phase=phase) + + def _push_pending_repo(self, repo: Repo, *, phase: str) -> bool: remote_oid = self._remote_oid() if remote_oid == self._pending_push_oid: pushed = True @@ -655,11 +664,12 @@ class StateGitSync: self._pending_manifest = None self.last_error = None self.healthy = True + self._last_success_monotonic = time.monotonic() self.log(f"[state-sync] generation={self.generation} phase={phase} status=ok") return True def sync(self, phase: str) -> bool: - with self._mutex: + with self._mutex, ExitStack() as resources: try: if self._workspace is None: raise StateSyncError("state workspace is not initialized") @@ -667,6 +677,13 @@ class StateGitSync: # another commit or generation for the same state transition. if self._pending_push_oid is not None: return self._push_pending(phase=phase) + # Intent/result WAL writes are never throttled. Idle/full + # cycles need not rewrite multi-megabyte state every 15s. + if (self.healthy and phase in {"cycle", "cycle_no_slots"} + and time.monotonic() - self._last_success_monotonic < 60): + return True + if self._refresh_workspace_pending: + self._refresh_local_workspace() previous_manifest: dict[str, Any] = {} try: previous_manifest = read_json(self._workspace / "manifest.json") @@ -691,7 +708,7 @@ class StateGitSync: "checksums": checksums, } write_json(self._workspace / "manifest.json", manifest) - repo = Repo(str(self._workspace)) + repo = resources.enter_context(Repo(str(self._workspace))) porcelain.add(repo) status = porcelain.status(repo) staged = any(status.staged.get(kind) for kind in ("add", "delete", "modify")) @@ -704,7 +721,7 @@ class StateGitSync: author=self._author, committer=self._author, ) - commit_count = sum(1 for _ in repo.get_walker()) + commit_count = sum(1 for _ in islice(repo.get_walker(), self.history_depth + 1)) if commit_count > self.history_depth: branch_ref = f"refs/heads/{self.branch}".encode("utf-8") del repo.refs[branch_ref] @@ -715,6 +732,7 @@ class StateGitSync: author=self._author, committer=self._author, ) + self._refresh_workspace_pending = True local_oid = repo.head().decode("ascii") self._pending_push_oid = local_oid self._pending_generation = next_generation @@ -726,6 +744,32 @@ class StateGitSync: self.log(f"[state-sync] phase={phase} status=failed reason={self.last_error}") return False + def _refresh_local_workspace(self) -> None: + """After a verified orphan push, discard unreachable *local* objects. + + Merely resetting the branch bounds history but leaves old blobs on + disk. Rotate only this class's disposable workspace after verification. + """ + old = self._workspace + parent = Path(tempfile.mkdtemp(prefix="modelhub-agent-state-")) + replacement = parent / "state" + try: + with porcelain.clone(self.remote, replacement, branch=self.branch, depth=1, + errstream=io.BytesIO(), **self._auth_kwargs()) as repo: + if repo.head().decode("ascii") != self._expected_remote_oid: + raise StateSyncError("state changed during local workspace rotation") + self._validate_manifest(replacement) + except Exception: + shutil.rmtree(parent) + raise + self._workspace = replacement + self._refresh_workspace_pending = False + if (old is not None and old.name == "state" + and old.parent.name.startswith("modelhub-agent-state-") + and old.parent.parent == Path(tempfile.gettempdir())): + shutil.rmtree(old.parent) + self.log("[state-sync] local_workspace=rotated unreachable_objects=discarded") + def write_readiness(self, *, ready: bool, reason: str | None = None, extra: dict[str, Any] | None = None) -> None: body = { "ready": bool(ready), diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 0d4309eb82..9d176a9515 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.10.18.1" +AGENT_VERSION = "2026.10.26.1" diff --git a/tests/test_takeover_accuracy.py b/tests/test_takeover_accuracy.py new file mode 100644 index 0000000000..647d17036b --- /dev/null +++ b/tests/test_takeover_accuracy.py @@ -0,0 +1,114 @@ +import copy +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "modelhub_submmit_api")) +from models import ModelInspection +from candidate_preflight import CandidatePreflightAdvisor +from profile_admission import filter_profile_batch, profile_key +from outcome_tracker import OutcomeTracker +from common import write_json +from state_sync import StateGitSync +from dulwich import porcelain + + +class TakeoverAccuracyTests(unittest.TestCase): + def tracker(self, root): + return OutcomeTracker(root / "outcomes.jsonl", checkpoint_path=root / "checkpoint.json", + recent_path=root / "recent.jsonl") + + def test_mlx_and_unknown_quantization_not_plain_weights(self): + for config, expected in [({"quantization": {"bits": 4, "group_size": 64}}, "mlx"), + ({"quantization_config": {"bits": 4}}, "unknown")]: + inspection = ModelInspection(repo_id="owner/model", model_config=config) + self.assertEqual(expected, inspection.quantization_method) + advisor = CandidatePreflightAdvisor() + result = advisor.assess(inspection=inspection, task_type="text-generation", + target_gpu="MetaX_c-500", framework="vllm", config_params="x") + self.assertFalse(result.allowed) + self.assertEqual("preflight_unverified_weight_format", result.reason) + self.assertEqual("mlx", ModelInspection(repo_id="m", model_card_metadata={ + "library_name": "mlx"}).quantization_method) + self.assertIsNone(ModelInspection(repo_id="m", model_config={}).quantization_method) + + def test_encoding_fingerprints_distinguish_compressed_formats(self): + a = ModelInspection(repo_id="a", model_config={"quantization_config": { + "quant_method": "compressed-tensors", "format": "float-quantized"}}) + b = ModelInspection(repo_id="b", model_config={"quantization_config": { + "quant_method": "compressed-tensors", "format": "pack-quantized"}}) + self.assertNotEqual(a.quantization_signature, b.quantization_signature) + + def candidate(self, model="a"): + return {"repoId": model, "targetGpu": "gpu", "framework": "vllm", "taskType": "text-generation", + "preflightMetadata": {"modelType": "family", "quantizationSignature": "awq:abc"}} + + def test_batch_guard_counts_pending_and_same_batch_and_is_retryable(self): + candidate = self.candidate() + pending = {**candidate, "outcome": "pending"} + selected, deferred = filter_profile_batch([self.candidate(str(i)) for i in range(4)], [pending], {}) + self.assertEqual(1, len(selected)) + self.assertEqual(3, len(deferred)) + selected, deferred = filter_profile_batch(deferred, [], {}) + self.assertEqual(2, len(selected)) + + def test_proven_encoding_releases_cap_but_failures_do_not(self): + candidates = [self.candidate(str(i)) for i in range(4)] + key = "|".join(profile_key(candidates[0])) + stats = {"encodingCombinationStats": {key: {"successCount": 3, "failureCount": 0}}} + self.assertEqual(4, len(filter_profile_batch(candidates, [], stats)[0])) + stats["encodingCombinationStats"][key]["failureCount"] = 10 + self.assertEqual(2, len(filter_profile_batch(candidates, [], stats)[0])) + + def test_takeover_is_idempotent_and_does_not_invent_version(self): + with tempfile.TemporaryDirectory() as tmp: + tracker = self.tracker(Path(tmp)) + tasks = [{"taskId": "old", "modelId": "owner/model", "gpuType": "gpu", "status": "waiting"}] + self.assertEqual(1, tracker.adopt_active_tasks(tasks)) + self.assertEqual(0, tracker.adopt_active_tasks(tasks)) + self.assertEqual(1, tracker.pending_route_count_for_model("owner/model")) + self.assertNotIn("agentVersion", tracker.pending_records()[0]) + + def test_encoding_and_version_statistics_survive_compaction(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + tracker = self.tracker(root) + for i in range(3): + tracker.record_submission("m"+str(i), "gpu", "vllm", "text-generation", str(i), + "2026-10-26T00:00:00+00:00", self.candidate()["preflightMetadata"]) + tracker._update_record_from_task(tracker._by_task_id[str(i)], {"status": "success", "verifyResult": 1}) + before = tracker.get_stats_report() + tracker._compact_if_needed(force=True) + restored = self.tracker(root).get_stats_report() + self.assertEqual(before["encodingCombinationStats"], restored["encodingCombinationStats"]) + self.assertEqual(before["agentSummaries"], restored["agentSummaries"]) + + def test_state_rotation_preserves_remote_and_bounds_local_objects(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + remote = root / "remote.git" + with porcelain.init(remote, bare=True): + pass + project = root / "project" + project.mkdir() + manager = StateGitSync(project_root=project, credentials={"username": "u", "email": "u@x", "password": "p"}, + remote=str(remote), history_depth=2, log_fn=lambda _: None) + try: + self.assertTrue(manager.restore()) + for i in range(5): + write_json(project / ".modelhub_state/account_capacity.json", {"generation": i}) + self.assertTrue(manager.sync("result"), manager.last_error) + self.assertTrue(manager.healthy) + self.assertEqual(manager._expected_remote_oid, manager._remote_oid()) + with patch.object(manager, "_copy_snapshot", side_effect=AssertionError("idle writes should throttle")): + self.assertTrue(manager.sync("cycle_no_slots")) + write_json(project / ".modelhub_state/account_capacity.json", {"generation": 99}) + self.assertTrue(manager.sync("intent")) + finally: + manager.close() + + +if __name__ == "__main__": + unittest.main()