fix: verify capacity from dual paginated active-task scans

Do not cache transient repository-list failures as empty weight sets.
Deduplicate normalized file paths without lowering reported file sizes.
Record whitelisted OOM evidence and bounded discovery diagnostics.

README.md

@@ -2,6 +2,104 @@
 
 This repository is packaged for the ModelHub XC agent platform.
 
+## Capacity pagination verification 2026.10.28.1
+
+Supersedes the 2026.10.27.1 policy of using the larger reported total.
+The platform can return inflated totals, incorrect page counts, and short pages.
+Capacity now reads waiting/running pages through an explicit empty page, ignores
+total/pages for termination, and deduplicates task IDs. It repeats the scan with
+page sizes 100 and 50; only matching active ID sets confirm occupancy. Reported
+totals are diagnostic only. Repeated pages, malformed records, request failures,
+changing snapshots or the 100-page-per-status budget fail closed for that account.
+This costs additional read requests but scans active tasks, not full history.
+Two agreeing reads are not an atomic snapshot or a guarantee against a platform
+omission shared by both queries; server submission limits remain authoritative.
+Logs report `method=dual_page_scan`, `confirmed_active`, and safe failure reasons.
+No age cleanup, routing, success-rate gates, recent-five-slot reserve, or weekly
+capacity exploration policy changes. Existing tasks are not cancelled.
+
+## Previous capacity refresh fix 2026.10.27.1
+
+All task-page requests are clamped to the platform's 100-record limit, including
+community sampling. Normal and weekly capacity refreshes now use the same
+filtered waiting/running pagination, validate responses and deduplicate task IDs.
+When reported totals disagree with returned rows, occupancy uses the larger
+value conservatively and logs the discrepancy. Missing or non-progressing pages
+fail closed. Failed reads report `active_counts=unknown` and
+`reason=capacity_read_unavailable`, retaining the last-success age for diagnosis;
+they do not advertise cached occupancy as a freshly confirmed full queue.
+Healthy accounts remain usable, and the next successful refresh reopens slots.
+The five-slot recent-model reserve remains in force.
+
+## 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
+`/opt/entrypoint.sh    - /model` executable error, even in restored cached
+official configurations. It preserves terminal compatibility errors when a
+log also contains transient platform resource messages. Framework scoring
+includes unresolved non-platform failures; sparse model-profile evidence is
+blended with eight prior observations instead of discarded below a threshold.
+
+For operation alongside the baseline agent, the new default state branch is
+`agent-state-accuracy-v2`. Do not override it with the baseline's `agent-state`.
+The separate branch starts its own checkpoint on first boot; it does not
+overwrite or migrate the baseline state. New submissions retain agent version,
+instance ID and platform strategy ID. Historical tasks are not assigned an
+invented version.
+
+`MODELHUB_COEXISTENCE_MODE=1` is the default: this agent does not run queue
+cancellation, including deterministic cleanup, because the accounts are shared
+with the baseline. Submission preflight remains enabled. Set this variable to
+`0` only when account-wide cleanup is intended; `--disable-queue-cleanup` still
+disables cleanup explicitly. The baseline agent is unchanged. Both agents share
+account capacity and platform deduplication; separate state does not reserve
+capacity or eliminate simultaneous submission races.
+
 ## Platform Contract
 
 - Root-level `Dockerfile`
@@ -67,7 +165,7 @@ Optional tuning:
 - `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` default `5` per account
 - `MODELHUB_RECENT_MODEL_DAYS` default `7`
 - `MODELHUB_STATE_SYNC_REMOTE` default `https://dev.modelhub.org.cn/CoolBoy/submmit.git`
-- `MODELHUB_STATE_SYNC_BRANCH` default `agent-state`
+- `MODELHUB_STATE_SYNC_BRANCH` default `agent-state-accuracy-v2`
 - `MODELHUB_STATE_SYNC_BATCH_SIZE` default `20`
 
 Git state synchronization reads `modelhub_user_name`, `modelhub_user_email`, and
@@ -140,7 +238,8 @@ unvetted GPU or framework.
 ## Durable State Branch
 
 The hosted worker synchronizes an allowlisted runtime snapshot to the orphan
-`agent-state` branch of this repository. Before each batch of at most 20 API
+`agent-state-accuracy-v2` branch of this repository (overridable via
+`MODELHUB_STATE_SYNC_BRANCH`). Before each batch of at most 20 API
 submissions it pushes write-ahead intents containing model, GPU, task, framework,
 safe parameter vector, and a config fingerprint. It pushes task IDs and results
 after the batch. If either push fails, no further submission begins until the
@@ -301,11 +400,43 @@ model/GPU combinations, the same run progressively expands discovery to the last
 7 days, the last 30 days, and finally older history (up to 3,000 models). Scanning
 stops as soon as enough replacement candidates have been found.
 
+When only recent-model reserved positions remain, discovery also deepens the
+same recent window up to the configured scan cap (3,000 by default). It never
+admits older models into those positions or lowers the quality gates. This avoids
+rechecking only an exhausted short prefix on every poll.
+
+An ordinary account-capacity rejection temporarily marks that account full and
+tries the next account. It does not lower the persisted capacity using a cached
+active count: another worker may have filled the queue since that count was
+read. Downward capacity learning remains confined to weekly calibration.
+
+Account allocation is age-aware: recent models first use accounts with only
+reserved positions left, then the account with the fewest shared positions
+remaining. Old models continue to use the account with the most eligible space.
+Largest-headroom and rotating tie-breaks balance otherwise equivalent accounts;
+weekly calibration remains pinned to its target account. For example, with
+accounts at 94/100 and 95/100 and five recent-only positions each, assigning a
+recent model to the second account preserves the first account's last old-model
+position. A recent-then-old pair can both be accepted rather than stranding the
+old model after the recent model takes its only eligible position.
+
 Model verification results are reused for 15 minutes across poll cycles, and a
 locally failed model/GPU pair cools down for 24 hours instead of being excluded
 forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done`
 includes `skip_reasons` so an empty candidate pool is directly diagnosable.
 
+Repository-tree HTTP failures are reported as retriable inspection failures,
+not cached as empty repositories. Repeated normalized file paths count once
+(using the largest reported size for that path), while different weight formats
+still contribute separately to the full-repository OOM gate.
+
+For an empty-candidate investigation, read `daily_runs/<cycle>/summary.json`,
+then follow `waveResults[].summary.runDir` to `skipped.jsonl` and `failed.jsonl`.
+OOM skips now include size/capacity/source evidence, and `scanStages` includes
+`discoveryQueries` with raw rows, accepted rows, cache hits, and stop reasons.
+Reaching a page/result budget does not mean the catalog is exhausted. These
+diagnostics contain selected fields only, never full configs or HTTP headers.
+
 Community deduplication is scoped to the exact model/GPU combination. A model
 adapted on one GPU remains eligible for another GPU. Before each submission the
 worker performs an uncached community check; a lookup failure defers the task

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

modelhub_submmit_api/candidate_preflight.py

@@ -189,14 +189,22 @@ 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,
+            "publishedSizeBytes": inspection.published_size_bytes,
             "gpuMemoryGiB": self.gpu_memory_gib.get(target_gpu),
             "gpuMemoryEvidence": self.gpu_memory_evidence.get(target_gpu),
         }
         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,
@@ -272,7 +280,9 @@ class CandidatePreflightAdvisor:
             required_gib = memory_sizing_bytes / (1024**3) * MODEL_LOAD_OVERHEAD
             metadata["estimatedRequiredGiB"] = round(required_gib, 3)
             metadata["memorySizingBasis"] = (
-                "recursive_repository_on_disk" if repository_bytes else "selected_weights_fallback"
+                "published_repository_upper_bound"
+                if (inspection.published_size_bytes or 0) > (inspection.exact_repository_size_bytes or 0)
+                else ("recursive_repository_on_disk" if repository_bytes else "selected_weights_fallback")
             )
             metadata["maximumRepositorySizeGiB"] = round(memory_gib / MODEL_LOAD_OVERHEAD, 3)
             if required_gib > memory_gib:

modelhub_submmit_api/failure_taxonomy.py

@@ -67,7 +67,7 @@ PLATFORM_PATTERNS = (
     (re.compile(r"welcome\.sh: No such file|data: command not found", re.I), "broken_platform_launch_script"),
     (re.compile(r"id_rsa.*No such file", re.I), "missing_platform_credential"),
     (re.compile(r"找不到空闲卡|no idle (?:gpu|card)", re.I), "no_idle_device"),
-    (re.compile(r"storage|download.*timed? out|connection reset", re.I), "platform_io_transient"),
+    (re.compile(r"no space left on device|download.*timed? out|connection reset", re.I), "platform_io_transient"),
 )
 
 DETERMINISTIC_LOG_PATTERNS = (
@@ -80,6 +80,8 @@ DETERMINISTIC_LOG_PATTERNS = (
 # semantics. Generic messages such as "attention backend not supported" or
 # "GPU type not supported" must not create an architecture blacklist.
 EXPLICIT_FRAMEWORK_MODEL_UNSUPPORTED_PATTERNS = (
+    re.compile(r"model type\s+[`'\"][^`'\"]+[`'\"].{0,100}does not recognize this architecture", re.I),
+    re.compile(r"Model architectures?\s*\[[^\]\n]+\]\s*(?:are|is) not supported", re.I),
     re.compile(
         r"(?:当前|该|此)(?:推理)?框架.{0,100}(?:不支持|暂不支持|无法支持|不兼容)"
         r".{0,100}(?:该|此|当前)?(?:模型架构|模型|架构)",
@@ -117,19 +119,16 @@ def classify_failure_report(report_code: str | None, log_lines: Iterable[str] =
     code = str(report_code or "").strip().upper()
     text = "\n".join(str(line) for line in log_lines)
 
-    # Infrastructure signatures override broad report codes such as
-    # EXECUTE_EMPTY_RESULT so they do not poison model compatibility feedback.
-    for pattern, reason in PLATFORM_PATTERNS:
-        if pattern.search(text):
-            return FailureClassification(
-                "platform_infrastructure",
-                "gpu_framework",
-                "open_short_gpu_framework_circuit_and_retry_other_models",
-                True,
-                False,
-                reason,
-            )
-
+    if re.search(r'exec:.*entrypoint.*(?:no such file|not found)', text, re.I):
+        return FailureClassification(
+            "configuration_launch", "configuration", "reject_broken_config", True, False,
+            "invalid_executable_argv",
+        )
+    if re.search(r'quantization method .*not supported for the current GPU', text, re.I):
+        return FailureClassification(
+            "quantization_compatibility", "model_gpu_framework", "reject_quantization_route",
+            True, False, "explicit_quantization_gpu_incompatibility",
+        )
     if code in DETERMINISTIC_POLICIES:
         return DETERMINISTIC_POLICIES[code]
     for pattern, inferred_code in DETERMINISTIC_LOG_PATTERNS:
@@ -149,6 +148,15 @@ def classify_failure_report(report_code: str | None, log_lines: Iterable[str] =
         )
     if code in SEMANTIC_POLICIES:
         return SEMANTIC_POLICIES[code]
+    # Fall back to infrastructure signatures only after specific root causes.
+    # Logs may include a transient allocation failure before a model starts.
+    for pattern, reason in PLATFORM_PATTERNS:
+        if pattern.search(text):
+            return FailureClassification(
+                "platform_infrastructure", "gpu_framework",
+                "open_short_gpu_framework_circuit_and_retry_other_models",
+                True, False, reason,
+            )
     return FailureClassification(
         "ambiguous_runtime",
         "unknown",

modelhub_submmit_api/hf_discovery.py

@@ -110,6 +110,7 @@ class HuggingFaceDiscovery:
         self._model_card_lock = threading.Lock()
         self._popular_page_lock = threading.Lock()
         self._popular_page_cursor = 0
+        self.last_listing_diagnostics: list[dict[str, Any]] = []
 
     @staticmethod
     def _remember_lru(cache: OrderedDict, key: Any, value: Any, limit: int) -> None:
@@ -133,6 +134,7 @@ class HuggingFaceDiscovery:
             return []
 
         deduped: dict[str, HFModelSummary] = {}
+        self.last_listing_diagnostics = []
         del read_concurrency
 
         per_tag_limit = limit if len(pipeline_tags) <= 1 else max(10, (limit + len(pipeline_tags) - 1) // len(pipeline_tags))
@@ -156,6 +158,7 @@ class HuggingFaceDiscovery:
                 updated_after=updated_after,
                 sort=sort,
                 start_page=start_page,
+                diagnostics=self.last_listing_diagnostics,
             ):
                 current = deduped.get(model.repo_id)
                 if current is None or (model.last_modified or parse_datetime("1970-01-01")) > (
@@ -175,6 +178,7 @@ class HuggingFaceDiscovery:
         updated_after=None,
         sort: str = "last_modified",
         start_page: int = 1,
+        diagnostics: list[dict[str, Any]] | None = None,
     ) -> list[HFModelSummary]:
         # ModelScope OpenAPI rejects values above 50 with InputParameterError.
         page_size = min(max(1, limit), 50)
@@ -183,12 +187,23 @@ class HuggingFaceDiscovery:
         filter_disabled = self._unsupported_task_filters.get(task_tag, 0.0) > time.monotonic()
         models: list[HFModelSummary] = []
         page_count = (max_items + page_size - 1) // page_size
+        diagnostic = {
+            "pipelineTag": pipeline_tag, "sort": sort,
+            "requestedLimit": limit, "effectiveLimit": max_items,
+            "pageSize": page_size, "startPage": start_page,
+            "pagesRead": 0, "cachedPages": 0, "rawRowsScanned": 0,
+            "acceptedModels": 0, "stopReason": "page_budget",
+            "catalogExhausted": False,
+        }
+        if diagnostics is not None:
+            diagnostics.append(diagnostic)
         for page_number in range(start_page, min(61, start_page + page_count)):
             cache_key = (sort, "*" if filter_disabled else task_tag, page_number, page_size)
             cached = self._model_page_cache.get(cache_key)
             if cached is not None and time.monotonic() - cached[0] < self._model_page_cache_ttl:
                 self._model_page_cache.move_to_end(cache_key)
                 items = list(cached[1])
+                diagnostic["cachedPages"] += 1
             else:
                 elapsed = time.monotonic() - self._last_model_page_request_at
                 if self._last_model_page_request_at > 0 and elapsed < self._page_interval_seconds:
@@ -211,6 +226,7 @@ class HuggingFaceDiscovery:
                     self._last_model_page_request_at = time.monotonic()
                     if exc.status_code == 400 and not filter_disabled:
                         self._unsupported_task_filters[task_tag] = time.monotonic() + 86_400
+                        diagnostic["stopReason"] = "unsupported_filter_retry"
                         print(
                             f"[modelscope] task_filter_unsupported task={task_tag} "
                             "fallback=unfiltered ttl=86400s",
@@ -223,12 +239,15 @@ class HuggingFaceDiscovery:
                             updated_after=updated_after,
                             sort=sort,
                             start_page=start_page,
+                            diagnostics=diagnostics,
                         )
                     print(
                         f"[modelscope] list_models_error task={task_tag} page={page_number} "
                         f"partial_models={len(models)} retry_next_cycle=true error={exc}",
                         flush=True,
                     )
+                    diagnostic["stopReason"] = "http_error"
+                    diagnostic["httpStatus"] = exc.status_code
                     break
 
                 items = self._extract_models(payload)
@@ -238,7 +257,11 @@ class HuggingFaceDiscovery:
                     (time.monotonic(), list(items)),
                     self._page_cache_limit,
                 )
+            diagnostic["pagesRead"] += 1
+            diagnostic["rawRowsScanned"] += len(items)
             if not items:
+                diagnostic["stopReason"] = "empty_page"
+                diagnostic["catalogExhausted"] = True
                 break
             for item in items:
                 model = self._parse_model(
@@ -249,9 +272,13 @@ class HuggingFaceDiscovery:
                 )
                 if model is not None:
                     models.append(model)
+                    diagnostic["acceptedModels"] += 1
                 if len(models) >= max_items:
+                    diagnostic["stopReason"] = "result_limit"
                     return models
             if len(items) < page_size:
+                diagnostic["stopReason"] = "short_page"
+                diagnostic["catalogExhausted"] = True
                 break
         return models
 
@@ -412,7 +439,10 @@ class HuggingFaceDiscovery:
             )
         except HttpJsonError as exc:
             print(f"[modelscope] repo_tree_error repo={repo_id} error={exc}", flush=True)
-            payload = None
+            # A failed lookup is not an empty repository. Let the caller defer
+            # inspection and retry next cycle instead of caching missing
+            # weights as a permanent framework-incompatibility decision.
+            raise
 
         entries = self._extract_files(payload)
 
@@ -570,7 +600,7 @@ def inspect_repo_tree(repo_id: str, entries: list[dict[str, Any]]) -> ModelInspe
                 break
         if size_present:
             try:
-                file_sizes[path] = max(0, int(size_value))
+                file_sizes[path] = max(file_sizes.get(path, 0), int(size_value))
             except (TypeError, ValueError):
                 pass
         filename = PurePosixPath(path).name.lower()
@@ -584,12 +614,12 @@ def inspect_repo_tree(repo_id: str, entries: list[dict[str, Any]]) -> ModelInspe
     selected_gguf = choose_best_gguf(gguf_files)
     return ModelInspection(
         repo_id=repo_id,
-        file_paths=sorted(file_paths),
+        file_paths=sorted(set(file_paths)),
         file_sizes=file_sizes,
-        gguf_files=sorted(gguf_files),
+        gguf_files=sorted(set(gguf_files)),
         selected_gguf=PurePosixPath(selected_gguf).name if selected_gguf else None,
-        weight_files=sorted(vllm_weight_files),
-        onnx_files=sorted(onnx_files),
+        weight_files=sorted(set(vllm_weight_files)),
+        onnx_files=sorted(set(onnx_files)),
     )
 
 

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:

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
@@ -377,6 +378,7 @@ def choose_candidate_for_gpu_detailed(
     require_official_config: bool = False,
     config_optimizer: SafeConfigOptimizer | None = None,
     excluded_frameworks: set[str] | None = None,
+    rejection_details: list[dict[str, Any]] | None = None,
 ) -> tuple[CandidateModel | None, str | None]:
     last_rejection_reason: str | None = None
     for task_type in task_types:
@@ -473,6 +475,23 @@ def choose_candidate_for_gpu_detailed(
                 )
                 if not assessment.allowed:
                     last_rejection_reason = assessment.reason
+                    if rejection_details is not None:
+                        rejection_details.append({
+                            "taskType": task_type,
+                            "framework": framework,
+                            "reason": assessment.reason,
+                            "metadata": {
+                                key: assessment.metadata[key]
+                                for key in (
+                                    "modelType", "architectures", "quantizationMethod",
+                                    "estimatedLoadBytes", "repositoryOnDiskBytes",
+                                    "recursiveRepositoryBytes", "publishedSizeBytes",
+                                    "gpuMemoryGiB", "gpuMemoryEvidence",
+                                    "estimatedRequiredGiB", "memorySizingBasis",
+                                    "maximumRepositorySizeGiB",
+                                ) if key in assessment.metadata
+                            },
+                        })
                     continue
                 config_params = assessment.config_params
                 warnings.extend(assessment.warnings)
@@ -651,6 +670,14 @@ def process_model_for_candidates(
                         "repoId": model.repo_id,
                         "targetGpu": target_gpu,
                         "reason": "preflight_predicted_oom",
+                        "preflightDetails": {
+                            "stage": "published_size_precheck",
+                            "publishedSizeBytes": model.file_size,
+                            "estimatedRequiredGiB": model.file_size / (1024**3) * MODEL_LOAD_OVERHEAD,
+                            "gpuMemoryGiB": memory_gib,
+                            "gpuMemoryEvidence": preflight_advisor.gpu_memory_evidence.get(target_gpu),
+                            "loadOverhead": MODEL_LOAD_OVERHEAD,
+                        },
                     })
                     continue
 
@@ -711,12 +738,15 @@ 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)
         for target_gpu, task_types, excluded_frameworks in pending_routes:
             if len(candidates) - before_count >= MAX_ROUTES_PER_MODEL:
                 break
+            rejection_details: list[dict[str, Any]] = []
             best, preflight_reason = choose_candidate_for_gpu_detailed(
                 model=model,
                 inspection=inspection,
@@ -728,6 +758,7 @@ def process_model_for_candidates(
                 require_official_config=official_registry is not None,
                 config_optimizer=config_optimizer,
                 excluded_frameworks=excluded_frameworks,
+                rejection_details=rejection_details,
             )
             if best is None:
                 reason = (
@@ -738,7 +769,10 @@ def process_model_for_candidates(
                         else "no_compatible_auto_template_or_framework"
                     )
                 )
-                skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason})
+                skipped.append({
+                    "repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason,
+                    **({"preflightRejections": rejection_details} if rejection_details else {}),
+                })
                 continue
             record = candidate_to_record(best)
             if excluded_frameworks:
@@ -822,6 +856,17 @@ def build_adaptive_scan_stages(
                 "limit": max_models,
             }
         )
+    if not allow_older_than_recent_window:
+        # Reserved positions still need a deep search of eligible recent
+        # models. Widening the date window alone repeatedly scans the same
+        # short prefix (often already processed) on every poll cycle.
+        stages.append(
+            {
+                "name": "recent_window_full",
+                "updatedAfter": recent_cutoff,
+                "limit": max_models,
+            }
+        )
     deduped: list[dict[str, Any]] = []
     seen: set[tuple[str | None, int, str]] = set()
     for stage in stages:
@@ -1620,6 +1665,7 @@ def run_submission(
             "candidateModelsAfterStage": candidate_model_count(candidates),
             "skippedAdded": len(skipped) - skipped_before_stage,
             "failedAdded": len(stage_failed),
+            "discoveryQueries": list(getattr(hf_discovery, "last_listing_diagnostics", [])),
         }
         scan_stages.append(stage_summary)
         print(
@@ -1660,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),
@@ -1678,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),

modelhub_submmit_api/market_intelligence.py

@@ -4,6 +4,7 @@ import math
 import re
 import statistics
 import threading
+import yaml
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta
 from pathlib import Path
@@ -97,6 +98,31 @@ def _validated_official_config(config: str, *, framework: str, target_gpu: str)
         return False, "missing_sut_or_ref_config"
     if any(marker in config for marker in ("{{", "}}", "PLACEHOLDER")):
         return False, "unresolved_placeholder"
+    try:
+        document = yaml.safe_load(config)
+    except yaml.YAMLError:
+        return False, "invalid_yaml"
+    if not isinstance(document, dict):
+        return False, "invalid_config_structure"
+
+    def valid_commands(node: Any) -> bool:
+        if isinstance(node, dict):
+            for key, value in node.items():
+                if key in {"command", "entrypoint"} and isinstance(value, list):
+                    if not value or not all(isinstance(arg, str) for arg in value):
+                        return False
+                    # argv[0] is an executable, not a shell command. In
+                    # particular, a folded YAML list item must not eat argv[1].
+                    if not value[0].strip() or re.search(r"\s", value[0]):
+                        return False
+                if not valid_commands(value):
+                    return False
+        elif isinstance(node, list):
+            return all(valid_commands(value) for value in node)
+        return True
+
+    if not valid_commands(document):
+        return False, "malformed_command_argv"
 
     declared = re.search(r"(?m)^\s*framework:\s*['\"]?([^'\"\s]+)", config)
     if declared is None or declared.group(1).strip() != framework:
@@ -384,7 +410,7 @@ class MarketIntelligenceManager:
         }
         payload = client.list_tasks_page(
             current=1,
-            page_size=200,
+            page_size=100,
             only_mine=False,
             begin_time=cutoff,
             end_time=now,
@@ -736,7 +762,7 @@ class MarketIntelligenceManager:
         local_success = max(0, int(local_item.get("successCount") or 0))
         local_failure = max(
             0,
-            int(local_item.get("attributableFailureCount", local_item.get("failureCount") or 0)),
+            int(local_item.get("failureCount") or 0) - int(local_item.get("platformFailureCount") or 0),
         )
         local_samples = local_success + local_failure
         local_score = _wilson_lower_bound(local_success, local_samples)
@@ -746,7 +772,7 @@ class MarketIntelligenceManager:
         recent_all_failures = max(0, int(recent_item.get("failureCount") or 0))
         recent_failure = max(
             0,
-            int(recent_item.get("attributableFailureCount", recent_item.get("failureCount") or 0)),
+            int(recent_item.get("failureCount") or 0) - int(recent_item.get("platformFailureCount") or 0),
         )
         recent_samples = recent_success + recent_failure
         recent_rate = recent_success / recent_samples if recent_samples else None
@@ -849,11 +875,31 @@ class MarketIntelligenceManager:
             if incumbent_best > 0 and float(evidence["combinedScore"]) < incumbent_best * NEW_FRAMEWORK_PROMOTION_MARGIN:
                 continue
             promoted.append(framework)
-        return self.rank_frameworks(
+        ranked = self.rank_frameworks(
             task_type=task_type,
             target_gpu=target_gpu,
             compatible_frameworks=promoted,
         )
+        model_type = str(getattr(inspection, "model_type", None) or "").strip()
+        quantization = str(getattr(inspection, "quantization_method", None) or "none").strip()
+        if not model_type:
+            return ranked
+
+        def profile_score(framework: str) -> float:
+            prior = float(self._framework_evidence(task_type, target_gpu, framework)["combinedScore"])
+            key = f"{target_gpu}|{framework}|{task_type}|{model_type}|{quantization}"
+            item = (self.local_outcome_stats.get("profileCombinationStats") or {}).get(key) or {}
+            success = max(0, int(item.get("successCount") or 0))
+            failure = max(0, int(item.get("failureCount") or 0) - int(item.get("platformFailureCount") or 0))
+            samples = success + failure
+            if not samples:
+                return prior
+            # Model-specific negative evidence must affect the framework
+            # choice, not only the ranking after the first framework wins.
+            lower = _wilson_lower_bound(success + 8 * prior, samples + 8)
+            return lower
+
+        return sorted(ranked, key=profile_score, reverse=True)
 
     def compatible_discovered_frameworks(
         self,
@@ -941,6 +987,11 @@ class MarketIntelligenceManager:
         config = item.get("officialConfig")
         if not isinstance(config, str) or not config:
             return None
+        valid, reason = _validated_official_config(config, framework=framework, target_gpu=target_gpu)
+        if not valid:
+            item["officialConfigValid"] = False
+            item["officialConfigError"] = reason
+            return None
         return _render_official_config(config, gguf_filename=gguf_filename)
 
     def framework_metadata(self, task_type: str, target_gpu: str, framework: str) -> dict[str, Any]:

modelhub_submmit_api/modelhub_client.py

@@ -151,7 +151,7 @@ class ModelHubClient:
             "/api/adapt/task/page",
             query={
                 "current": current,
-                "pageSize": page_size,
+                "pageSize": max(1, min(100, int(page_size))),
                 "onlyMine": str(only_mine).lower(),
                 "beginTime": query_time(begin_time),
                 "endTime": query_time(end_time),
@@ -325,15 +325,63 @@ class ModelHubClient:
         return active_count
 
     def count_active_tasks_exact(self) -> int:
-        """Use the platform's filtered totals instead of paging all history."""
-        total = 0
-        for status in ("waiting", "running"):
-            page = self.list_tasks_page(current=1, page_size=1, only_mine=True, status=status)
-            data = page.get("data")
-            if not isinstance(data, dict) or "total" not in data:
-                raise ModelHubAPIError(f"Active-task total is unavailable for status={status}")
-            total += max(0, int(data["total"]))
-        return total
+        """Confirm active IDs with two independently paginated filtered reads.
+
+        The platform's total/pages fields can describe a different population.
+        Only an explicit empty page terminates a scan, never those fields or a
+        short page. Two page sizes must agree before advertising any free slots.
+        This is not an atomic snapshot: concurrent queue changes fail closed.
+        """
+        def invalid(reason: str) -> ModelHubAPIError:
+            error = ModelHubAPIError(reason)
+            error.capacity_reason = reason
+            return error
+
+        def scan(page_size: int) -> tuple[set[str], list[dict[str, Any]]]:
+            active: set[str] = set()
+            details = []
+            for status in ("waiting", "running"):
+                ids: set[str] = set()
+                reported = None
+                for current in range(1, 101):
+                    response = self.list_tasks_page(
+                        current=current, page_size=page_size, only_mine=True, status=status,
+                    )
+                    data = response.get("data") if isinstance(response, dict) else None
+                    if not isinstance(data, dict) or not isinstance(data.get("records"), list):
+                        raise invalid("invalid_active_response")
+                    try:
+                        value = int(data["total"])
+                        if value >= 0:
+                            reported = max(reported or 0, value)
+                    except (KeyError, ValueError, TypeError):
+                        pass  # Metadata is diagnostic, not an occupancy source.
+                    rows = data["records"]
+                    if not rows:
+                        break
+                    if len(rows) > page_size:
+                        raise invalid("page_size_not_respected")
+                    previous = len(ids)
+                    for row in rows:
+                        if not isinstance(row, dict) or not row.get("taskId") or row.get("status") != status:
+                            raise invalid("invalid_active_record")
+                        ids.add(str(row["taskId"]))
+                    if len(ids) == previous:
+                        raise invalid("active_pagination_no_progress")
+                else:
+                    raise invalid("active_pagination_budget_exhausted")
+                active.update(ids)
+                details.append({"status": status, "reported": reported, "unique": len(ids),
+                                "mismatch": reported != len(ids), "pageSize": page_size})
+            return active, details
+
+        self.active_count_details = []
+        first, details = scan(100)
+        confirmed, confirmation_details = scan(50)
+        if first != confirmed:
+            raise invalid("active_snapshot_changed")
+        self.active_count_details = details + confirmation_details
+        return len(confirmed)
 
     def _request(
         self,
@@ -521,6 +569,7 @@ class ModelHubClientPool:
         # persisted capacity is not enough without a successful active-count
         # read for the current scheduling window.
         self._count_known: list[bool] = [False for _ in clients]
+        self._count_success_at = [0.0 for _ in clients]
         self._active_refresh_at: float = 0.0
         self._counts_initialized = False
         self._state_lock = threading.Lock()
@@ -703,10 +752,14 @@ class ModelHubClientPool:
 
     def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int | None) -> int | None:
         try:
-            if max_count is None and hasattr(client, "count_active_tasks_exact"):
+            if hasattr(client, "count_active_tasks_exact"):
                 return client.count_active_tasks_exact()
-            return client.count_active_tasks(max_count=max_count, page_size=200)
-        except Exception:
+            return client.count_active_tasks(max_count=max_count, page_size=100)
+        except Exception as exc:
+            # Never log request URLs/headers or credentials.
+            code = getattr(exc, "code", None)
+            reason = getattr(exc, "capacity_reason", "request_failed")
+            print(f"[capacity] refresh_failed error={type(exc).__name__} code={code} reason={reason} count=unknown", flush=True)
             return None
 
     def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]:
@@ -779,6 +832,11 @@ class ModelHubClientPool:
                     old_remote_count = self._remote_counts[index]
                     new_remote_count = max(0, int(count))
                     self._count_known[index] = True
+                    self._count_success_at[index] = refreshed_at
+                    details = getattr(self.clients[index], "active_count_details", [])
+                    if any(item.get("mismatch") for item in details):
+                        print(f"[capacity] account={index + 1:02d} total_records_mismatch=true "
+                              f"confirmed_active={new_remote_count} method=dual_page_scan details={details}", flush=True)
                     if new_remote_count > self._account_caps[index]:
                         self._account_caps[index] = new_remote_count
                         caps_changed = True
@@ -821,6 +879,7 @@ class ModelHubClientPool:
             1
             for index in range(len(self.clients))
             if index not in self._capacity_probe_attempted
+            and self._count_known[index]
             and self._capacity_probe_cycle >= self._capacity_probe_cooldown_until[index]
             and self._effective_count_locked(index) >= self._account_caps[index]
         )
@@ -846,9 +905,20 @@ class ModelHubClientPool:
             normal_slots = sum(
                 max(0, self._account_caps[index] - self._effective_count_locked(index))
                 for index in range(len(self.clients))
+                if self._count_known[index]
             )
             return normal_slots if normal_slots > 0 else self._probe_slot_count_locked()
 
+    def active_count_status(self) -> list[dict[str, Any]]:
+        self._refresh_active_counts()
+        with self._state_lock:
+            now = time.monotonic()
+            return [{"account": i + 1, "known": self._count_known[i],
+                     "active": self._effective_count_locked(i) if self._count_known[i] else None,
+                     "lastSuccessAgeSeconds": round(now - self._count_success_at[i], 1)
+                     if self._count_success_at[i] else None}
+                    for i in range(len(self.clients))]
+
     def list_tasks(self, **kwargs):  # noqa: ANN003, ANN001
         # Default: use only the reader client to avoid fanout amplification.
         # The fanout_all parameter allows explicit cross-account merging when needed.
@@ -935,6 +1005,7 @@ class ModelHubClientPool:
                 index: self._account_caps[index] - self._effective_count_locked(index)
                 for index in range(len(self.clients))
                 if index not in excluded
+                and self._count_known[index]
                 and (weekly_target is None or index == weekly_target)
                 and (
                     reserve_capacity_slots is None
@@ -948,6 +1019,26 @@ class ModelHubClientPool:
             usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0}
             if not usable:
                 return None
+            if (
+                reserve_capacity_slots is None
+                and weekly_target is None
+                and self.recent_model_reserve_slots > 0
+            ):
+                # Recent models can use every position; old models cannot.
+                # Pack recent work into reserved positions first, then the
+                # account nearest its old-model threshold. This preserves
+                # shared positions for later old candidates, regardless of
+                # candidate arrival order. Keep largest-headroom/round-robin
+                # balancing among accounts with the same shared capacity.
+                shared_slots = {
+                    index: max(0, remaining - self.recent_model_reserve_slots)
+                    for index, remaining in usable.items()
+                }
+                least_shared = min(shared_slots.values())
+                usable = {
+                    index: remaining for index, remaining in usable.items()
+                    if shared_slots[index] == least_shared
+                }
             best_remaining = max(usable.values())
             tied = [index for index, remaining in usable.items() if remaining == best_remaining]
             selected_index = min(
@@ -977,6 +1068,7 @@ class ModelHubClientPool:
                 index
                 for index in range(len(self.clients))
                 if index not in excluded
+                and self._count_known[index]
                 and (weekly_target is None or index == weekly_target)
                 and (weekly_target is not None or index not in self._capacity_probe_attempted)
                 and (
@@ -1029,7 +1121,12 @@ class ModelHubClientPool:
                 self._capacity_probe_cooldown_until[index] = (
                     self._capacity_probe_cycle + self._capacity_probe_cooldown_cycles
                 )
-            else:
+            elif self._weekly_target_locked() == index:
+                # Only the scheduled calibration may lower a learned cap.
+                # Ordinary rejection can race with another agent filling this
+                # account; our cached count is not evidence of its capacity.
+                # Persisting it (even zero -> one) strands otherwise usable
+                # positions until the next calibration.
                 observed_capacity = max(1, self._effective_count_locked(index))
                 if observed_capacity < self._account_caps[index]:
                     self._account_caps[index] = observed_capacity

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 (
@@ -140,12 +163,17 @@ class ModelInspection:
         return None
 
     @property
-    def repository_size_bytes(self) -> int | None:
+    def exact_repository_size_bytes(self) -> int | None:
         """Return exact recursive on-disk size when every file has a size."""
-        exact: int | None = None
         if self.file_paths and not any(path not in self.file_sizes for path in self.file_paths):
             total = sum(max(0, int(self.file_sizes[path])) for path in self.file_paths)
-            exact = total if total > 0 else None
+            return total if total > 0 else None
+        return None
+
+    @property
+    def repository_size_bytes(self) -> int | None:
+        """Use the conservative maximum of recursive and published sizes."""
+        exact = self.exact_repository_size_bytes
         published = int(self.published_size_bytes or 0) or None
         values = [value for value in (exact, published) if value is not None]
         return max(values) if values else None

modelhub_submmit_api/outcome_tracker.py

@@ -20,8 +20,10 @@ 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
 
 
 DEFAULT_OUTCOMES_PATH = Path("outcomes/submissions.jsonl")
@@ -36,6 +38,7 @@ FAILURE_ENRICHMENT_MAX_ATTEMPTS = 3
 TERMINAL_TASK_STATUSES = {"success", "failed", "error", "cancelled", "completed"}
 
 DECISION_RECORD_FIELDS = {
+    "agentVersion", "instanceId", "strategyId",
     "taskId",
     "modelId",
     "targetGpu",
@@ -325,6 +328,9 @@ class OutcomeTracker:
         model_profile: dict[str, Any] | None = None,
     ) -> None:
         record: dict[str, Any] = {
+            "agentVersion": AGENT_VERSION,
+            "instanceId": runtime_instance_id(),
+            "strategyId": os.getenv("STRATEGY_ID"),
             "modelId": model_id,
             "targetGpu": target_gpu,
             "framework": framework,
@@ -355,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
@@ -754,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"
@@ -770,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):
@@ -908,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,
@@ -938,6 +976,8 @@ class OutcomeTracker:
             "frameworkSummaries",
             "combinationStats",
             "profileCombinationStats",
+            "encodingCombinationStats",
+            "agentSummaries",
             "sizedProfileCombinationStats",
         ):
             base_items = baseline.get(field) if isinstance(baseline.get(field), dict) else {}

modelhub_submmit_api/poll_runner.py

@@ -234,7 +234,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--post-cycle-cooldown-seconds", type=int, default=2, help="Short sleep after a successful cycle")
     parser.add_argument("--max-cycles", type=int, default=0, help="Optional hard stop after N cycles; 0 means run until quota is reached")
     parser.add_argument("--print-stats", action="store_true", help="Load outcomes, sync, print stats report, and exit")
-    parser.add_argument("--disable-queue-cleanup", action="store_true", help=argparse.SUPPRESS)
+    parser.add_argument("--disable-queue-cleanup", action="store_true",
+                        default=os.getenv("MODELHUB_COEXISTENCE_MODE", "1") != "0",
+                        help=argparse.SUPPRESS)
     parser.add_argument(
         "--queue-cleanup-interval-cycles",
         type=int,
@@ -666,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 = (
@@ -696,12 +699,19 @@ 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
     STATS_PRINT_INTERVAL = 10
 
     log(f"[poll] version={AGENT_VERSION} poll_run_dir={poll_run_dir}")
+    log(
+        f"[coexistence] state_branch={base_args.state_sync_branch} "
+        f"queue_cancellation={'disabled' if base_args.disable_queue_cleanup else 'enabled'} "
+        "submission_preflight=enabled"
+    )
     log(
         "[state-retention] mode=decision_state_only full_archive=disabled "
         f"recent_outcomes={DEFAULT_RECENT_OUTCOME_LIMIT} "
@@ -882,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(
@@ -1040,8 +1056,12 @@ def run_poll_loop(
             if hasattr(modelhub_client, "weekly_calibration_status") else {}
         )
         available_slots = modelhub_client.available_submit_slots() if hasattr(modelhub_client, "available_submit_slots") else None
+        count_status = modelhub_client.active_count_status() if hasattr(modelhub_client, "active_count_status") else []
+        display_counts = [str(item["active"]) if item["known"] else "unknown" for item in count_status]
+        if count_status and any(not item["known"] for item in count_status):
+            log(f"[capacity] account_status={count_status} unknown_accounts=deferred")
         log(
-            f"[poll] cycle={cycles} active_counts={','.join(str(count) for count in active_counts) if active_counts else 'n/a'} "
+            f"[poll] cycle={cycles} active_counts={','.join(display_counts or [str(count) for count in active_counts]) or 'n/a'} "
             f"capacity_limits={','.join(str(limit) for limit in capacity_limits) if capacity_limits else 'n/a'} "
             f"capacity_probe={'on' if capacity_probe else 'off'} "
             f"weekly_calibration={'on' if calibration.get('targetAccount') else 'off'} "
@@ -1068,7 +1088,8 @@ def run_poll_loop(
                     state_sync.healthy = False
                     state_sync.last_error = str(exc)
                     state_sync.write_readiness(ready=False, reason="state_sync_unhealthy")
-            log(f"[poll] cycle={cycles} sleep={base_args.poll_interval_seconds}s reason=no_available_slots")
+            no_slot_reason = "capacity_read_unavailable" if any(not item["known"] for item in count_status) else "no_available_slots"
+            log(f"[poll] cycle={cycles} sleep={base_args.poll_interval_seconds}s reason={no_slot_reason}")
             time.sleep(base_args.poll_interval_seconds)
             continue
 

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

modelhub_submmit_api/routing_engine.py

@@ -46,6 +46,13 @@ class SuccessFirstRoutingEngine:
         return value
 
     def _evidence(self, candidate: dict[str, Any]) -> tuple[str, int, int, float]:
+        def observed_total(item: dict[str, Any]) -> int:
+            if "failureCount" in item:
+                return max(0, int(item.get("successCount") or 0) + int(item.get("failureCount") or 0)
+                           - int(item.get("platformFailureCount") or 0))
+            return max(0, int(item.get("decisionTotal") or 0))
+
+        sparse = None
         profile = candidate.get("preflightMetadata") or {}
         gpu = str(candidate.get("targetGpu") or "")
         framework = str(candidate.get("framework") or "")
@@ -60,14 +67,18 @@ class SuccessFirstRoutingEngine:
             size_bucket = int(load_bytes).bit_length() - 1
             key = f"{gpu}|{framework}|{task_type}|{model_type}|{quantization}|{size_bucket}"
             item = (self.outcome_stats.get("sizedProfileCombinationStats") or {}).get(key) or {}
-            total = int(item.get("decisionTotal") or 0)
+            total = observed_total(item)
+            if total:
+                sparse = ("local_profile_size_shrunk", item, total)
             if total >= 20:
                 success = int(item.get("successCount") or 0)
                 return "local_profile_size", success, total, wilson_lower_bound(success, total)
         if model_type:
             key = f"{gpu}|{framework}|{task_type}|{model_type}|{quantization}"
             item = (self.outcome_stats.get("profileCombinationStats") or {}).get(key) or {}
-            total = int(item.get("decisionTotal") or 0)
+            total = observed_total(item)
+            if total and sparse is None:
+                sparse = ("local_profile_shrunk", item, total)
             if total >= 40:
                 success = int(item.get("successCount") or 0)
                 return "local_profile", success, total, wilson_lower_bound(success, total)
@@ -97,6 +108,13 @@ class SuccessFirstRoutingEngine:
             total = 2
             success = 1
             level = "global_conservative_prior"
+        if sparse is not None:
+            level, item, samples = sparse
+            # Eight prior observations stabilize small groups without hiding
+            # their failures behind an unreachable hard sample threshold.
+            prior_success = round(8 * success / max(1, total))
+            success = int(item.get("successCount") or 0) + prior_success
+            total = samples + 8
         return level, success, total, wilson_lower_bound(success, total)
 
     def annotate(self, candidate: dict[str, Any]) -> dict[str, Any]:
@@ -171,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 ""),

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
@@ -17,14 +20,14 @@ from typing import Any, Iterable
 from dulwich import porcelain
 from dulwich.repo import Repo
 
-from common import read_json, read_jsonl, write_json, write_jsonl
+from common import read_json, read_jsonl, write_json, write_jsonl, runtime_instance_id
 from runner_common import load_key_files
 from version import AGENT_VERSION
 
 
 STATE_SCHEMA_VERSION = 1
 DEFAULT_REMOTE = "https://dev.modelhub.org.cn/CoolBoy/submmit.git"
-DEFAULT_BRANCH = "agent-state"
+DEFAULT_BRANCH = "agent-state-accuracy-v2"
 DEFAULT_BATCH_SIZE = 100
 DEFAULT_HISTORY_DEPTH = 200
 DEFAULT_RECENT_TERMINAL_INTENTS = 300
@@ -32,6 +35,7 @@ DEFAULT_LEDGER_RECORDS = 500
 DEFAULT_CRASH_RECORDS = 50
 
 STATE_OUTCOME_FIELDS = {
+    "agentVersion", "instanceId", "strategyId",
     "taskId", "modelId", "targetGpu", "framework", "taskType", "submitTime",
     "lastSyncTime", "status", "verifyResult", "outcome", "failReason",
     "modelProfile", "failureCode", "failureCategory", "failureScope",
@@ -170,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:
@@ -235,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,
@@ -244,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
 
@@ -267,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)
@@ -276,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)
@@ -342,6 +349,7 @@ class StateGitSync:
         retained_terminal = terminal[:DEFAULT_RECENT_TERMINAL_INTENTS]
         discarded = terminal[DEFAULT_RECENT_TERMINAL_INTENTS:]
         compact_fields = {
+            "agentVersion", "instanceId", "strategyId",
             "intentId", "batchId", "status", "createdAt", "completedAt",
             "repoId", "targetGpu", "taskType", "framework", "configSource",
             "configFingerprint", "safeConfigVector", "taskId",
@@ -372,6 +380,9 @@ class StateGitSync:
         config = str(candidate.get("configParams") or "")
         return {
             "intentId": uuid.uuid4().hex,
+            "agentVersion": AGENT_VERSION,
+            "instanceId": runtime_instance_id(),
+            "strategyId": os.getenv("STRATEGY_ID"),
             "batchId": batch_id,
             "status": "pending",
             "createdAt": _utc_now().isoformat(),
@@ -622,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
@@ -650,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")
@@ -662,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")
@@ -686,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"))
@@ -699,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]
@@ -710,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
@@ -721,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),

modelhub_submmit_api/version.py

@@ -1 +1 @@
-AGENT_VERSION = "2026.09.22.1"
+AGENT_VERSION = "2026.10.28.1"

tests/test_accuracy_audit.py

@@ -0,0 +1,96 @@
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+from types import SimpleNamespace
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "modelhub_submmit_api"))
+from failure_taxonomy import classify_failure_report
+from market_intelligence import _validated_official_config, MarketIntelligenceManager
+from routing_engine import SuccessFirstRoutingEngine
+from state_sync import StateGitSync, DEFAULT_BRANCH
+from poll_runner import build_parser
+
+
+class AccuracyAuditTests(unittest.TestCase):
+    def config(self, command):
+        return f"framework: vllm_fix_tokenizer\nsut_config:\n  command: {command}\nref_config: {{}}\n"
+
+    def test_broken_official_and_cached_config_rejected(self):
+        bad = self.config('["/opt/entrypoint.sh    - /model", "--port", "80"]')
+        for gpu in ("Biren_166m", "Kunlunxin_p-800"):
+            self.assertEqual((False, "malformed_command_argv"), _validated_official_config(
+                bad, framework="vllm_fix_tokenizer", target_gpu=gpu))
+            with tempfile.TemporaryDirectory() as tmp:
+                manager = MarketIntelligenceManager(Path(tmp) / "market.json")
+                manager.state = {"frameworkStats": {"text-generation": {gpu: {
+                    "vllm_fix_tokenizer": {"officialConfigValid": True, "officialConfig": bad}}}}}
+                self.assertIsNone(manager.official_config(task_type="text-generation", target_gpu=gpu,
+                                                        framework="vllm_fix_tokenizer"))
+
+    def test_separate_argv_and_shell_command_remain_valid(self):
+        for command in ('["/opt/entrypoint.sh", "/model"]', '["/bin/sh", "-c", "echo hello"]'):
+            self.assertEqual((True, None), _validated_official_config(
+                self.config(command), framework="vllm_fix_tokenizer", target_gpu="Biren_166m"))
+
+    def test_specific_failure_survives_transient_resource_error(self):
+        result = classify_failure_report("MODEL_NOT_SUPPORTED", [
+            "找不到空闲卡", "当前框架版本不支持该模型架构"])
+        self.assertEqual("framework_architecture_unsupported", result.category)
+        self.assertEqual("tokenizer_compatibility", classify_failure_report(
+            "TOKENIZER_FAILED", ["no idle gpu", "Tokenizer class missing"]).category)
+        self.assertEqual("platform_infrastructure", classify_failure_report(
+            "EXECUTE_EMPTY_RESULT", ["no idle gpu"]).category)
+        self.assertEqual("framework_architecture_unsupported", classify_failure_report(
+            None, ["no idle gpu", "Model architectures ['NewForCausalLM'] are not supported for now"]).category)
+
+    def test_launch_and_quantization_have_specific_categories(self):
+        self.assertEqual("configuration_launch", classify_failure_report("EXECUTE_EMPTY_RESULT", [
+            'exec: "/opt/entrypoint.sh    - /model": no such file or directory']).category)
+        self.assertEqual("quantization_compatibility", classify_failure_report("EXECUTE_EMPTY_RESULT", [
+            "The quantization method compressed-tensors is not supported for the current GPU"]).category)
+
+    def test_sparse_failures_lower_score_even_without_classification(self):
+        candidate = {"targetGpu": "gpu", "framework": "vllm", "taskType": "text-generation",
+                     "frameworkMarketSamples": 1000, "frameworkMarketSuccessRate": 0.8,
+                     "preflightMetadata": {"modelType": "family", "estimatedLoadBytes": 1024}}
+        with tempfile.TemporaryDirectory() as tmp:
+            engine = SuccessFirstRoutingEngine(Path(tmp) / "route.json")
+            prior = engine._evidence(candidate)[3]
+            engine.outcome_stats = {"profileCombinationStats": {
+                "gpu|vllm|text-generation|family|none": {
+                    "successCount": 0, "failureCount": 5, "decisionTotal": 0}}}
+            evidence = engine._evidence(candidate)
+            self.assertEqual("local_profile_shrunk", evidence[0])
+            self.assertLess(evidence[3], prior)
+
+    def test_coexistence_defaults_and_submission_provenance(self):
+        with patch.dict(os.environ, {"MODELHUB_COEXISTENCE_MODE": "1", "STRATEGY_ID": "new-agent"}):
+            self.assertTrue(build_parser().parse_args([]).disable_queue_cleanup)
+            intent = StateGitSync._intent({"repoId": "owner/model"}, "batch")
+            self.assertEqual("new-agent", intent["strategyId"])
+            self.assertTrue(intent["agentVersion"])
+            self.assertTrue(intent["instanceId"])
+        self.assertNotEqual("agent-state", DEFAULT_BRANCH)
+
+    def test_model_specific_failures_change_framework_selection(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            manager = MarketIntelligenceManager(Path(tmp) / "market.json")
+            manager.state = {"frameworkStats": {"text-generation": {"gpu": {
+                "first": {"modelCount": 1000, "wilsonLowerBound": 0.8},
+                "second": {"modelCount": 1000, "wilsonLowerBound": 0.6},
+            }}}}
+            manager.set_local_outcome_stats({"profileCombinationStats": {
+                "gpu|first|text-generation|family|none": {"successCount": 0, "failureCount": 4}
+            }})
+            with patch.object(manager, "compatible_discovered_frameworks", return_value=[]):
+                ranked = manager.selectable_frameworks(
+                    task_type="text-generation", target_gpu="gpu", incumbent_frameworks=["first", "second"],
+                    inspection=SimpleNamespace(model_type="family", quantization_method=None))
+            self.assertEqual(["second", "first"], ranked)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_candidate_preflight.py

@@ -243,6 +243,26 @@ class CandidatePreflightTests(unittest.TestCase):
         self.assertFalse(assessment.allowed)
         self.assertEqual("recursive_repository_on_disk", assessment.metadata["memorySizingBasis"])
 
+    def test_memory_diagnostics_distinguish_published_upper_bound(self) -> None:
+        gib = 1024 ** 3
+        inspection = ModelInspection(
+            repo_id="owner/published-larger",
+            file_paths=["config.json", "tokenizer.json", "model.safetensors"],
+            file_sizes={"config.json": 0, "tokenizer.json": 0, "model.safetensors": 20 * gib},
+            weight_files=["model.safetensors"],
+            model_config={"model_type": "qwen2"},
+            published_size_bytes=40 * gib,
+        )
+        assessment = CandidatePreflightAdvisor(gpu_memory_gib={"gpu": 32}).assess(
+            inspection=inspection, task_type="text-generation", target_gpu="gpu",
+            framework="vllm", config_params="max_model_len: 4096",
+        )
+        self.assertEqual("preflight_predicted_oom", assessment.reason)
+        self.assertEqual(20 * gib, assessment.metadata["recursiveRepositoryBytes"])
+        self.assertEqual(40 * gib, assessment.metadata["publishedSizeBytes"])
+        self.assertEqual(48.0, assessment.metadata["estimatedRequiredGiB"])
+        self.assertEqual("published_repository_upper_bound", assessment.metadata["memorySizingBasis"])
+
     def test_p800_verified_capacity_accepts_80_gib_repository_boundary(self) -> None:
         gib = 1024**3
 

tests/test_capacity_refresh.py

@@ -0,0 +1,122 @@
+import sys
+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 modelhub_client import ModelHubClient, ModelHubClientPool, ModelHubAPIError
+
+
+class CapacityRefreshTests(unittest.TestCase):
+    def test_api_boundary_clamps_all_callers(self):
+        client = ModelHubClient(token="test")
+        with patch.object(client, "_request", return_value={}) as request:
+            client.list_tasks_page(page_size=200)
+            self.assertEqual(100, request.call_args.kwargs["query"]["pageSize"])
+
+    def client(self, total=97, observed=94):
+        client = ModelHubClient(token="test")
+        def page(*, status, page_size, current=1, **kwargs):
+            self.assertLessEqual(page_size, 100)
+            return {"data": {"total": total if status == "waiting" else 0, "records": [
+                {"taskId": str(i), "status": status} for i in range(
+                    (current - 1) * page_size, min(current * page_size, observed if status == "waiting" else 0))]}}
+        client.list_tasks_page = page
+        return client
+
+    def test_full_cache_refreshes_to_free_slots_with_disagreeing_total(self):
+        client = self.client()
+        pool = ModelHubClientPool([client], active_task_cap=100)
+        pool._remote_counts[0] = 100
+        self.assertEqual([94], pool.refresh_active_counts())
+        self.assertEqual(6, pool.available_submit_slots())
+        self.assertTrue(client.active_count_details[0]["mismatch"])
+
+    def test_error_is_unknown_and_recovery_reopens_slots(self):
+        client = self.client()
+        pool = ModelHubClientPool([client], active_task_cap=100)
+        pool.refresh_active_counts()
+        with patch.object(client, "list_tasks_page", side_effect=ModelHubAPIError("分页大小不能超过100")):
+            pool.refresh_active_counts()
+            self.assertEqual(0, pool.available_submit_slots())
+            self.assertIsNone(pool.active_count_status()[0]["active"])
+            self.assertIsNone(pool._reserve_account(set()))
+        pool.refresh_active_counts()
+        self.assertEqual(6, pool.available_submit_slots())
+
+    def test_incomplete_or_repeated_pages_do_not_open_slots(self):
+        client = self.client(total=200, observed=100)
+        client.list_tasks_page = lambda **kwargs: {"data": {"records": [
+            {"taskId": "same", "status": kwargs["status"]}]}}
+        with self.assertRaises(ModelHubAPIError):
+            client.count_active_tasks_exact()
+        client.list_tasks_page = lambda **kwargs: {"data": {}}
+        with self.assertRaises(ModelHubAPIError):
+            client.count_active_tasks_exact()
+
+    def test_duplicate_ids_and_transition_are_counted_once(self):
+        client = ModelHubClient(token="test")
+        client.list_tasks_page = lambda **kwargs: {"data": {"total": 1, "records": [] if kwargs['current'] > 1 else [
+            {"taskId": "same", "status": kwargs["status"]},
+            {"taskId": "same", "status": kwargs["status"]}]}}
+        self.assertEqual(1, client.count_active_tasks_exact())
+
+    def test_total_lower_than_observed_uses_larger_count(self):
+        self.assertEqual(99, self.client(total=97, observed=99).count_active_tasks_exact())
+
+    def test_inflated_total_does_not_hide_capacity(self):
+        self.assertEqual(100, self.client(total=773, observed=100).count_active_tasks_exact())
+
+    def test_underreported_pages_do_not_truncate_scan(self):
+        client = self.client(total=58, observed=175)
+        original = client.list_tasks_page
+        def page(**kwargs):
+            result = original(**kwargs)
+            result['data']['pages'] = 1
+            return result
+        client.list_tasks_page = page
+        self.assertEqual(175, client.count_active_tasks_exact())
+
+    def test_different_confirmation_ids_fail_closed_even_with_equal_counts(self):
+        client = self.client(total=1, observed=1)
+        original = client.list_tasks_page
+        def page(**kwargs):
+            result = original(**kwargs)
+            if kwargs['page_size'] == 50:
+                for row in result['data']['records']:
+                    row['taskId'] = 'replacement'
+            return result
+        client.list_tasks_page = page
+        with self.assertRaisesRegex(ModelHubAPIError, 'active_snapshot_changed'):
+            client.count_active_tasks_exact()
+
+    def test_confirmation_error_does_not_open_slots(self):
+        client = self.client(total=773, observed=94)
+        original = client.list_tasks_page
+        def page(**kwargs):
+            if kwargs['page_size'] == 50:
+                raise ModelHubAPIError('temporary failure')
+            return original(**kwargs)
+        client.list_tasks_page = page
+        pool = ModelHubClientPool([client], active_task_cap=100)
+        pool.refresh_active_counts()
+        self.assertEqual(0, pool.available_submit_slots())
+        self.assertFalse(pool.active_count_status()[0]['known'])
+
+    def test_empty_queue_confirmed_despite_stale_total(self):
+        self.assertEqual(0, self.client(total=773, observed=0).count_active_tasks_exact())
+
+    def test_short_page_is_not_end_of_scan(self):
+        client = ModelHubClient(token='test')
+        def page(**kwargs):
+            rows = [{'taskId': str(kwargs['current']), 'status': 'waiting'}] if (
+                kwargs['status'] == 'waiting' and kwargs['current'] <= 3) else []
+            return {'data': {'records': rows}}
+        client.list_tasks_page = page
+        self.assertEqual(3, client.count_active_tasks_exact())
+
+    def test_scan_budget_is_bounded(self):
+        client = ModelHubClient(token='test')
+        client.list_tasks_page = lambda **kw: {'data': {'records': [
+            {'taskId': str(kw['current']), 'status': kw['status']}]}}
+        with self.assertRaisesRegex(ModelHubAPIError, 'active_pagination_budget_exhausted'):
+            client.count_active_tasks_exact()

tests/test_concurrency.py

@@ -6,6 +6,7 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from unittest.mock import patch
@@ -216,17 +217,90 @@ def make_candidate(index: int) -> dict:
 
 
 class ClientPoolConcurrencyTests(unittest.TestCase):
+    def test_recent_work_preserves_shared_slots_for_later_old_models(self) -> None:
+        for initial_counts, ages, expected_counts in (
+            ([94, 95], ["recent", "old"], [95, 96]),
+            ([93, 94], ["recent", "recent", "old", "old"], [95, 96]),
+        ):
+            with self.subTest(initial_counts=initial_counts):
+                clients = [FakeClient(active_count=count) for count in initial_counts]
+                pool = ModelHubClientPool(
+                    clients,  # type: ignore[arg-type]
+                    active_task_cap=100,
+                    recent_model_reserve_slots=5,
+                    active_counts_ttl=60,
+                    capacity_probe_interval_cycles=0,
+                )
+                for index, age in enumerate(ages):
+                    pool.add_task_for_model(
+                        {"model": f"model-{index}"},
+                        model_last_modified=(
+                            "2026-10-01T00:00:00Z" if age == "recent"
+                            else "2025-01-01T00:00:00Z"
+                        ),
+                        submitted_at="2026-10-02T00:00:00Z",
+                    )
+                self.assertEqual(expected_counts, pool.active_task_counts())
+                self.assertEqual(len(ages), sum(len(client.submitted) for client in clients))
+                self.assertEqual(0, pool.old_model_submit_slots())
+
+    def test_recent_packing_keeps_largest_headroom_without_reserved_slots(self) -> None:
+        clients = [FakeClient(active_count=2), FakeClient(active_count=3)]
+        pool = ModelHubClientPool(
+            clients,  # type: ignore[arg-type]
+            active_task_cap=5,
+            recent_model_reserve_slots=0,
+        )
+        pool.add_task({"model": "recent"})
+        self.assertEqual([1, 0], [len(client.submitted) for client in clients])
+
+    def test_stale_capacity_rejection_does_not_poison_learned_limit(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            state_path = Path(temporary_dir) / "capacity.json"
+            client = FakeClient(active_count=0, reject_capacity=True)
+            pool = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                recent_model_reserve_slots=5,
+                capacity_probe_interval_cycles=0,
+                capacity_state_path=state_path,
+            )
+            self.assertEqual(100, pool.available_submit_slots())
+            with self.assertRaises(ModelHubAPIError):
+                pool.add_task({"model": "rejected-during-race"})
+
+            self.assertEqual([100], pool.account_capacity_limits())
+            client.reject_capacity = False
+            pool.refresh_active_counts()
+            self.assertEqual(95, pool.old_model_submit_slots())
+            pool.add_task_for_model(
+                {"model": "eligible-old-model"},
+                model_last_modified="2025-01-01T00:00:00Z",
+                submitted_at="2026-10-02T00:00:00Z",
+            )
+            self.assertEqual(1, len(client.submitted))
+            restarted = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                capacity_state_path=state_path,
+            )
+            self.assertEqual([100], restarted.account_capacity_limits())
+
     def test_exact_active_count_uses_filtered_platform_totals(self) -> None:
         client = ModelHubClient(token="test-token")
         calls: list[str] = []
 
-        def page(*, status: str, **_kwargs) -> dict:  # noqa: ANN003
+        def page(*, status: str, current: int = 1, page_size: int = 100, **_kwargs) -> dict:  # noqa: ANN003
             calls.append(status)
-            return {"data": {"total": {"waiting": 175, "running": 25}[status]}}
+            total = {"waiting": 175, "running": 25}[status]
+            return {"data": {"total": total, "records": [
+                {"taskId": f"{status}-{i}", "status": status}
+                for i in range((current - 1) * page_size, min(current * page_size, total))]}}
 
         client.list_tasks_page = page  # type: ignore[method-assign]
         self.assertEqual(200, client.count_active_tasks_exact())
-        self.assertEqual(["waiting", "running"], calls)
+        self.assertEqual(8, calls.count("waiting"))
+        self.assertEqual(4, calls.count("running"))
 
     @staticmethod
     def monday_calibration_time(hour: int = 10) -> datetime:
@@ -481,8 +555,37 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
             recent_model_days=7,
         )
 
-        self.assertEqual(["configured_window", "last_7_days"], [stage["name"] for stage in stages])
+        self.assertEqual(
+            ["configured_window", "last_7_days", "recent_window_full"],
+            [stage["name"] for stage in stages],
+        )
         self.assertTrue(all(stage["updatedAfter"] >= now - timedelta(days=7) for stage in stages))
+        self.assertEqual(3000, stages[-1]["limit"])
+
+    def test_recent_only_scan_deepens_even_when_date_window_cannot_expand(self) -> None:
+        now = datetime(2026, 8, 11, tzinfo=timezone.utc)
+        for initial_cutoff in (None, now - timedelta(days=7), now - timedelta(days=30)):
+            with self.subTest(initial_cutoff=initial_cutoff):
+                stages = build_adaptive_scan_stages(
+                    now=now,
+                    initial_updated_after=initial_cutoff,
+                    initial_limit=20,
+                    max_models=800,
+                    allow_older_than_recent_window=False,
+                )
+                self.assertEqual([20, 800], [stage["limit"] for stage in stages])
+                self.assertTrue(all(stage["updatedAfter"] == now - timedelta(days=7) for stage in stages))
+
+    def test_recent_only_scan_deduplicates_an_already_full_window(self) -> None:
+        now = datetime(2026, 8, 11, tzinfo=timezone.utc)
+        stages = build_adaptive_scan_stages(
+            now=now,
+            initial_updated_after=now - timedelta(days=7),
+            initial_limit=100,
+            max_models=100,
+            allow_older_than_recent_window=False,
+        )
+        self.assertEqual(1, len(stages))
 
     def test_submit_candidate_reports_old_model_policy_skip_at_dynamic_threshold(self) -> None:
         client = FakeClient(active_count=95)
@@ -738,6 +841,54 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
             self.assertEqual("last_modified", summary["scanStages"][-1]["sort"])
             self.assertIsNone(discovery.calls[-1])
 
+    def test_reserved_slots_find_recent_candidates_beyond_first_scan_prefix(self) -> None:
+        class RecentOnlyClient(ExhaustedRecentClient):
+            def old_model_submit_slots(self) -> int:
+                return 0
+
+        class PagedRecentDiscovery(FakeDiscovery):
+            def __init__(self) -> None:
+                super().__init__(4)
+                self.models[0] = replace(self.models[0], repo_id="owner/recent-exhausted")
+                self.limits: list[int] = []
+
+            def list_recent_models(self, *, limit, **_kwargs):  # noqa: ANN003
+                self.limits.append(limit)
+                return self.models[:limit]
+
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            root = Path(temporary_dir)
+            args = build_parser().parse_args([
+                "--task-types", "text-generation", "--since-hours", "168",
+                "--limit", "1", "--max-scan-models", "4",
+                "--skip-outcome-sync", "--skip-history-archive",
+                "--disable-gpu-strategy",
+            ])
+            for field, filename in (
+                ("runs_dir", "runs"), ("ledger_path", "ledger.jsonl"),
+                ("outcomes_path", "outcomes.jsonl"), ("claims_path", "claims.jsonl"),
+                ("history_archive_path", "history.jsonl"),
+                ("submission_exclusions_path", "exclusions.jsonl"),
+            ):
+                setattr(args, field, str(root / filename))
+            selector = TemplateSelector()
+            discovery = PagedRecentDiscovery()
+            client = RecentOnlyClient(
+                available=2,
+                processed_gpus=set(selector.supported_target_gpus("text-generation", auto_only=True)),
+            )
+            summary = run_submission(
+                args,
+                now=datetime(2026, 1, 2, 12, tzinfo=timezone.utc),
+                hf_discovery=discovery,  # type: ignore[arg-type]
+                modelhub_client=client,  # type: ignore[arg-type]
+                template_selector=selector,
+            )
+            self.assertEqual([1, 4], discovery.limits)
+            self.assertEqual(2, summary["submittedCount"])
+            self.assertEqual("recent_window_full", summary["scanStages"][-1]["name"])
+            self.assertEqual(4, summary["scannedModels"])
+
     def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None:
         with tempfile.TemporaryDirectory() as temporary_dir:
             root = Path(temporary_dir)

tests/test_market_intelligence.py

@@ -283,6 +283,7 @@ class MarketIntelligenceTests(unittest.TestCase):
                         "successCount": 10,
                         "failureCount": 3,
                         "attributableFailureCount": 0,
+                        "platformFailureCount": 3,
                     }
                 },
                 "recentCombinationStats": {
@@ -290,6 +291,7 @@ class MarketIntelligenceTests(unittest.TestCase):
                         "successCount": 10,
                         "failureCount": 3,
                         "attributableFailureCount": 0,
+                        "platformFailureCount": 3,
                         "consecutiveFailures": 0,
                         "consecutivePlatformFailures": 3,
                         "lastPlatformFailureAt": datetime.now(timezone.utc).isoformat(),

tests/test_modelscope_discovery.py

@@ -11,7 +11,7 @@ if str(PACKAGE_DIR) in sys.path:
     sys.path.remove(str(PACKAGE_DIR))
 sys.path.insert(0, str(PACKAGE_DIR))
 
-from hf_discovery import HuggingFaceDiscovery  # noqa: E402
+from hf_discovery import HuggingFaceDiscovery, inspect_repo_tree  # noqa: E402
 from http_json import HttpJsonError  # noqa: E402
 
 
@@ -58,6 +58,56 @@ class ModelMetadataClient:
 
 
 class ModelScopeDiscoveryTests(unittest.TestCase):
+    def test_listing_diagnostics_distinguish_filtered_prefix_from_exhaustion(self) -> None:
+        class FilteredPageClient:
+            def request_json(self, *_args, **_kwargs):
+                return {"data": {"models": [
+                    {"id": f"owner/model-{index}", "downloads": 0}
+                    for index in range(50)
+                ]}}
+
+        discovery = HuggingFaceDiscovery(http_client=FilteredPageClient())  # type: ignore[arg-type]
+        self.assertEqual([], discovery.list_recent_models(
+            pipeline_tags=["text-generation"], limit=50, min_downloads=50,
+        ))
+        diagnostic = discovery.last_listing_diagnostics[0]
+        self.assertEqual(50, diagnostic["rawRowsScanned"])
+        self.assertEqual(0, diagnostic["acceptedModels"])
+        self.assertEqual("page_budget", diagnostic["stopReason"])
+        self.assertFalse(diagnostic["catalogExhausted"])
+
+    def test_failed_repo_tree_is_retried_instead_of_cached_as_empty(self) -> None:
+        class RecoveringClient:
+            calls = 0
+
+            def request_json(self, *_args, **_kwargs):
+                self.calls += 1
+                if self.calls == 1:
+                    raise HttpJsonError("temporary outage", status_code=503)
+                return {"Data": {"Files": [{"Path": "model.safetensors", "Size": 1024}]}}
+
+        client = RecoveringClient()
+        discovery = HuggingFaceDiscovery(legacy_http_client=client)  # type: ignore[arg-type]
+        with self.assertRaises(HttpJsonError):
+            discovery.list_repo_tree("owner/recoverable")
+        self.assertNotIn("owner/recoverable", discovery._repo_tree_cache)
+        entries = discovery.list_repo_tree("owner/recoverable")
+        self.assertEqual("model.safetensors", entries[0]["Path"])
+        self.assertEqual(entries, discovery.list_repo_tree("owner/recoverable"))
+        self.assertEqual(2, client.calls)
+
+    def test_duplicate_normalized_paths_do_not_inflate_repository_size(self) -> None:
+        gib = 1024 ** 3
+        inspection = inspect_repo_tree("owner/model", [
+            {"Path": "model.safetensors", "Size": 20 * gib},
+            {"Path": "./model.safetensors", "Size": 19 * gib},
+            {"Path": "/model.safetensors", "Size": 20 * gib},
+        ])
+        self.assertEqual(["model.safetensors"], inspection.file_paths)
+        self.assertEqual(["model.safetensors"], inspection.weight_files)
+        self.assertEqual(20 * gib, inspection.repository_size_bytes)
+        self.assertEqual(20 * gib, inspection.estimated_load_bytes("vllm"))
+
     def test_model_last_modified_uses_detail_api_and_is_cached(self) -> None:
         metadata_client = ModelMetadataClient()
         discovery = HuggingFaceDiscovery(legacy_http_client=metadata_client)  # type: ignore[arg-type]

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()