2 Commits

Author SHA1 Message Date
CoolBoy
484a10607c fix: reject broken configs and improve accuracy feedback with isolated agent state 2026-10-18 16:29:18 +08:00
coolboy
964362818e fix: retry failed model inspections and preserve rejection evidence
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.
2026-10-03 01:37:55 +08:00
16 changed files with 408 additions and 37 deletions

View File

@@ -2,6 +2,30 @@
This repository is packaged for the ModelHub XC agent platform.
## 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 +91,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 +164,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
@@ -326,6 +351,18 @@ 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

View File

@@ -191,6 +191,8 @@ class CandidatePreflightAdvisor:
"quantizationMethod": inspection.quantization_method,
"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),
}
@@ -272,7 +274,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:

View File

@@ -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):
if re.search(r'exec:.*entrypoint.*(?:no such file|not found)', text, re.I):
return FailureClassification(
"platform_infrastructure",
"gpu_framework",
"open_short_gpu_framework_circuit_and_retry_other_models",
True,
False,
reason,
"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",

View File

@@ -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)),
)

View File

@@ -377,6 +377,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 +474,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 +669,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
@@ -717,6 +743,7 @@ def process_model_for_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 +755,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 +766,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:
@@ -1631,6 +1662,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(

View File

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

View File

@@ -140,12 +140,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

View File

@@ -22,6 +22,8 @@ from history_stats import classify_failure, is_failure, is_success
from llm_classifier import LLMAssistedClassifier
from modelhub_client import ModelHubClient, ModelHubClientPool
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,

View File

@@ -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,
@@ -702,6 +704,11 @@ def run_poll_loop(
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} "

View File

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

View File

@@ -17,14 +17,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 +32,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",
@@ -342,6 +343,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 +374,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(),

View File

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

View File

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

View File

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

View File

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

View File

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