From 5b1ec4d3eb9e5ee74f33da4327b42b54dfd1edb9 Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Sat, 22 Aug 2026 14:15:25 +0800 Subject: [PATCH] feat: add tiered durable state and memory bounds --- Dockerfile | 2 + README.md | 25 +- modelhub_submmit_api/config_optimizer.py | 9 +- modelhub_submmit_api/hf_discovery.py | 74 +++++- modelhub_submmit_api/main.py | 82 +++++- modelhub_submmit_api/outcome_tracker.py | 322 ++++++++++++++++++++++- modelhub_submmit_api/poll_runner.py | 27 +- modelhub_submmit_api/state_sync.py | 105 +++++++- modelhub_submmit_api/version.py | 2 +- tests/test_modelscope_discovery.py | 12 + tests/test_super_agent.py | 102 +++++++ 11 files changed, 722 insertions(+), 40 deletions(-) diff --git a/Dockerfile b/Dockerfile index 2ce290cb..118a524c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -2,6 +2,8 @@ FROM modelhubxc-4pd.tencentcloudcr.com/xc_agent_platform/python:3.11-slim ENV PYTHONUNBUFFERED=1 ENV PYTHONPATH=/app/modelhub_submmit_api +ENV NO_PROXY=dev.modelhub.org.cn,modelhub.org.cn +ENV no_proxy=dev.modelhub.org.cn,modelhub.org.cn WORKDIR /app diff --git a/README.md b/README.md index 1cfe1d09..c25aa458 100644 --- a/README.md +++ b/README.md @@ -375,11 +375,32 @@ probe and `/ready` reports worker availability. Poll and cleanup summaries are bounded in memory, and the last 200 worker crash records are synchronized with the durable state branch for post-restart diagnosis. +Version `2026.08.22.1` adds tiered durable outcome storage and hard runtime memory +bounds. The live `agent-state` branch contains only pending outcomes, an +aggregate checkpoint, and the most recent terminal window; completed raw +outcomes are sanitized, gzip-compressed, and uploaded to monthly +`agent-archive-YYYY-MM` branches. A restart restores only the live branch, so it +retains lifetime success statistics and architecture evidence without cloning or +parsing the full history. The first upgrade automatically compacts a legacy +outcome file after 2,000 records; later compactions are incremental. + +Candidate preparation now retains at most 100 distinct models (up to three +routes per model) per cycle by default. ModelScope repository-tree, config, +model-card, modification-time, and page caches use bounded LRU eviction, which +prevents memory use from growing with every poll cycle. These limits can be +tuned with `--candidate-pool-limit`, `MODELHUB_AGENT_OUTCOME_COMPACT_THRESHOLD`, +`MODELHUB_AGENT_RECENT_OUTCOME_LIMIT`, +`MODELSCOPE_DETAIL_CACHE_MAX_MODELS`, and `MODELSCOPE_PAGE_CACHE_MAX_PAGES`. +Cold archive upload is best-effort and retried on later state synchronizations; +the aggregate checkpoint remains the authoritative recovery input even while an +archive shard is awaiting upload. ModelHub Git traffic is excluded from proxy +routing in the container to avoid deployment-specific Git failures. + ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag -a agent-v28 -m "ModelHub agent 2026.08.21.1" -git push origin main agent-v28 +git tag -a agent-v29 -m "ModelHub agent 2026.08.22.1" +git push origin main agent-v29 ``` diff --git a/modelhub_submmit_api/config_optimizer.py b/modelhub_submmit_api/config_optimizer.py index d4c08c83..c1d9c32a 100644 --- a/modelhub_submmit_api/config_optimizer.py +++ b/modelhub_submmit_api/config_optimizer.py @@ -18,7 +18,13 @@ class SafeConfigOptimizer: self._qualified = self._learn() def _learn(self) -> dict[tuple[str, str, str], list[dict[str, Any]]]: - outcomes = read_jsonl(self.outcomes_path) + recent_path = Path(".modelhub_state/recent_outcomes.jsonl") + outcomes_by_key: dict[str, dict[str, Any]] = {} + for item in [*read_jsonl(recent_path), *read_jsonl(self.outcomes_path)]: + task_id = str(item.get("taskId") or "") + if task_id: + outcomes_by_key[task_id] = item + outcomes = list(outcomes_by_key.values()) by_task = { str(item.get("taskId")): item for item in outcomes @@ -117,4 +123,3 @@ class SafeConfigOptimizer: "evidenceLowerBound": choice["lowerBound"], "fingerprint": choice["fingerprint"], } - diff --git a/modelhub_submmit_api/hf_discovery.py b/modelhub_submmit_api/hf_discovery.py index f17e236d..87c07de6 100644 --- a/modelhub_submmit_api/hf_discovery.py +++ b/modelhub_submmit_api/hf_discovery.py @@ -4,6 +4,7 @@ import os import re import threading import time +from collections import OrderedDict from dataclasses import replace from datetime import datetime, timezone from pathlib import PurePosixPath @@ -70,13 +71,23 @@ class HuggingFaceDiscovery: timeout=timeout, retries=retries, ) - self._repo_tree_cache: dict[str, list[dict[str, Any]]] = {} + self._detail_cache_limit = max( + 32, + min(1000, int(os.getenv("MODELSCOPE_DETAIL_CACHE_MAX_MODELS", "128"))), + ) + self._page_cache_limit = max( + 16, + min(500, int(os.getenv("MODELSCOPE_PAGE_CACHE_MAX_PAGES", "96"))), + ) + self._repo_tree_cache: OrderedDict[str, list[dict[str, Any]]] = OrderedDict() self._repo_tree_lock = threading.Lock() - self._model_config_cache: dict[str, tuple[dict[str, Any], str | None]] = {} + self._model_config_cache: OrderedDict[str, tuple[dict[str, Any], str | None]] = OrderedDict() self._model_config_lock = threading.Lock() - self._model_last_modified_cache: dict[str, datetime | None] = {} + self._model_last_modified_cache: OrderedDict[str, datetime | None] = OrderedDict() self._model_last_modified_lock = threading.Lock() - self._model_page_cache: dict[tuple[str, int, int], tuple[float, list[dict[str, Any]]]] = {} + self._model_page_cache: OrderedDict[ + tuple[str, int, int], tuple[float, list[dict[str, Any]]] + ] = OrderedDict() self._model_page_cache_ttl = max( 0.0, float( @@ -95,9 +106,16 @@ class HuggingFaceDiscovery: ) self._last_model_page_request_at = 0.0 self._unsupported_task_filters: dict[str, float] = {} - self._model_card_cache: dict[str, dict[str, Any]] = {} + self._model_card_cache: OrderedDict[str, dict[str, Any]] = OrderedDict() self._model_card_lock = threading.Lock() + @staticmethod + def _remember_lru(cache: OrderedDict, key: Any, value: Any, limit: int) -> None: + cache[key] = value + cache.move_to_end(key) + while len(cache) > limit: + cache.popitem(last=False) + def list_recent_models( self, *, @@ -149,6 +167,7 @@ class HuggingFaceDiscovery: cache_key = ("*" 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]) else: elapsed = time.monotonic() - self._last_model_page_request_at @@ -191,7 +210,12 @@ class HuggingFaceDiscovery: break items = self._extract_models(payload) - self._model_page_cache[cache_key] = (time.monotonic(), list(items)) + self._remember_lru( + self._model_page_cache, + cache_key, + (time.monotonic(), list(items)), + self._page_cache_limit, + ) if not items: break for item in items: @@ -247,6 +271,8 @@ class HuggingFaceDiscovery: def get_model_card_metadata(self, repo_id: str) -> dict[str, Any]: with self._model_card_lock: cached = self._model_card_cache.get(repo_id) + if cached is not None: + self._model_card_cache.move_to_end(repo_id) if cached is not None: return dict(cached) encoded_repo_id = "/".join(quote(part, safe="") for part in repo_id.split("/")) @@ -262,12 +288,19 @@ class HuggingFaceDiscovery: except Exception: result = {} with self._model_card_lock: - self._model_card_cache[repo_id] = dict(result) + self._remember_lru( + self._model_card_cache, + repo_id, + dict(result), + self._detail_cache_limit, + ) return result def get_model_config(self, repo_id: str) -> tuple[dict[str, Any], str | None]: with self._model_config_lock: cached = self._model_config_cache.get(repo_id) + if cached is not None: + self._model_config_cache.move_to_end(repo_id) if cached is not None: return dict(cached[0]), cached[1] @@ -284,13 +317,20 @@ class HuggingFaceDiscovery: result = ({}, f"{type(exc).__name__}: {exc}") with self._model_config_lock: - self._model_config_cache[repo_id] = result + self._remember_lru( + self._model_config_cache, + repo_id, + result, + self._detail_cache_limit, + ) return dict(result[0]), result[1] def get_model_last_modified(self, repo_id: str) -> datetime | None: with self._model_last_modified_lock: if repo_id in self._model_last_modified_cache: - return self._model_last_modified_cache[repo_id] + result = self._model_last_modified_cache[repo_id] + self._model_last_modified_cache.move_to_end(repo_id) + return result encoded_repo_id = "/".join(quote(part, safe="") for part in repo_id.split("/")) try: @@ -325,12 +365,19 @@ class HuggingFaceDiscovery: result = parse_datetime(raw_value) with self._model_last_modified_lock: - self._model_last_modified_cache[repo_id] = result + self._remember_lru( + self._model_last_modified_cache, + repo_id, + result, + self._detail_cache_limit, + ) return result def list_repo_tree(self, repo_id: str) -> list[dict[str, Any]]: with self._repo_tree_lock: cached = self._repo_tree_cache.get(repo_id) + if cached is not None: + self._repo_tree_cache.move_to_end(repo_id) if cached is not None: return list(cached) @@ -348,7 +395,12 @@ class HuggingFaceDiscovery: entries = self._extract_files(payload) with self._repo_tree_lock: - self._repo_tree_cache[repo_id] = list(entries) + self._remember_lru( + self._repo_tree_cache, + repo_id, + list(entries), + self._detail_cache_limit, + ) return list(entries) @staticmethod diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index 3d59fe9c..5fb4c239 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -79,6 +79,12 @@ def build_parser() -> argparse.ArgumentParser: default=4, help="Multiplier used when auto-deriving scan limit from quota/queue capacity", ) + parser.add_argument( + "--candidate-pool-limit", + type=int, + default=100, + help="Maximum prepared candidates retained before submitting; bounds memory in large queues", + ) parser.add_argument("--min-downloads", type=int, default=50, help="Minimum downloads threshold") parser.add_argument("--daily-target", type=int, default=0, help="Daily submission target across all auto runs; 0 means unlimited") parser.add_argument("--dry-run", action="store_true", help="Only write artifacts without creating tasks") @@ -746,6 +752,7 @@ def collect_candidates_from_models( allow_dynamic_tasks: bool = False, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], int]: candidates: list[dict[str, Any]] = [] + candidate_models: set[str] = set() skipped: list[dict[str, Any]] = [] failed: list[dict[str, Any]] = [] processed_count = 0 @@ -754,7 +761,7 @@ def collect_candidates_from_models( new_models = [model for model in models if model.repo_id not in seen_model_ids] for offset in range(0, len(new_models), chunk_size): - if len(candidates) >= candidate_goal: + if len(candidate_models) >= candidate_goal: break chunk = new_models[offset : offset + chunk_size] for model in chunk: @@ -794,6 +801,10 @@ def collect_candidates_from_models( for index in range(len(chunk)): model_candidates, model_skipped, model_failed = ordered_results.get(index, ([], [], [])) candidates.extend(model_candidates) + if model_candidates: + model_id = str(model_candidates[0].get("repoId") or model_candidates[0].get("modelAddress") or "") + if model_id: + candidate_models.add(model_id) skipped.extend(model_skipped) failed.extend(model_failed) processed_count += len(chunk) @@ -927,6 +938,46 @@ def one_candidate_per_model(candidates: list[dict[str, Any]]) -> list[dict[str, return selected +def limit_routes_per_model( + candidates: list[dict[str, Any]], + *, + max_routes: int = 3, +) -> list[dict[str, Any]]: + selected: list[dict[str, Any]] = [] + counts: Counter[str] = Counter() + for candidate in candidates: + model_id = str(candidate.get("repoId") or candidate.get("modelAddress") or "") + if not model_id or counts[model_id] >= max(1, int(max_routes)): + continue + counts[model_id] += 1 + selected.append(candidate) + return selected + + +def candidate_model_count(candidates: list[dict[str, Any]]) -> int: + return len( + { + str(candidate.get("repoId") or candidate.get("modelAddress") or "") + for candidate in candidates + if candidate.get("repoId") or candidate.get("modelAddress") + } + ) + + +def limit_candidate_models(candidates: list[dict[str, Any]], *, max_models: int) -> list[dict[str, Any]]: + selected: list[dict[str, Any]] = [] + models: set[str] = set() + for candidate in candidates: + model_id = str(candidate.get("repoId") or candidate.get("modelAddress") or "") + if not model_id: + continue + if model_id not in models and len(models) >= max(0, int(max_models)): + continue + models.add(model_id) + selected.append(candidate) + return selected + + def run_submission( args: argparse.Namespace, *, @@ -1271,7 +1322,14 @@ def run_submission( if strategy_manager is not None: submission_goal = min(submission_goal, strategy_manager.submissions_until_refresh) attempt_multiplier = max(1, int(getattr(args, "scan_multiplier", 4) or 1)) - candidate_goal = max(submission_goal, submission_goal * attempt_multiplier) + raw_candidate_goal = max(submission_goal, submission_goal * attempt_multiplier) + candidate_pool_limit = max(20, int(getattr(args, "candidate_pool_limit", 100) or 100)) + candidate_goal = min(raw_candidate_goal, candidate_pool_limit) + print( + f"[scan] candidate_pool_limit={candidate_pool_limit} " + f"raw_goal={raw_candidate_goal} effective_goal={candidate_goal}", + flush=True, + ) pipeline_tags = pipeline_tags_for_task_types(selected_task_types) if official_registry is not None and not getattr(args, "task_types", None): @@ -1305,7 +1363,8 @@ def run_submission( allow_older_than_recent_window=allow_older_models_for_scan, recent_model_days=recent_model_days, ): - if candidate_goal <= 0 or len(candidates) >= candidate_goal: + current_candidate_models = candidate_model_count(candidates) + if candidate_goal <= 0 or current_candidate_models >= candidate_goal: break skipped_before_stage = len(skipped) stage_updated_after = stage["updatedAfter"] @@ -1348,7 +1407,7 @@ def run_submission( stage_candidates, stage_skipped, stage_failed, processed_count = collect_candidates_from_models( models=models, seen_model_ids=seen_model_ids, - candidate_goal=max(1, candidate_goal - len(candidates)), + candidate_goal=max(1, candidate_goal - current_candidate_models), hf_discovery=hf_discovery, modelhub_client=modelhub_client, template_selector=template_selector, @@ -1363,6 +1422,12 @@ def run_submission( config_optimizer=config_optimizer, allow_dynamic_tasks=not bool(getattr(args, "task_types", None)), ) + stage_candidates = limit_routes_per_model(stage_candidates, max_routes=3) + remaining_candidate_capacity = max(0, candidate_goal - current_candidate_models) + stage_candidates = limit_candidate_models( + stage_candidates, + max_models=remaining_candidate_capacity, + ) candidates.extend(stage_candidates) skipped.extend(stage_skipped) failed.extend(stage_failed) @@ -1375,13 +1440,15 @@ def run_submission( "newModelsProcessed": processed_count, "candidatesAdded": len(stage_candidates), "candidateCountAfterStage": len(candidates), + "candidateModelsAfterStage": candidate_model_count(candidates), "skippedAdded": len(skipped) - skipped_before_stage, "failedAdded": len(stage_failed), } scan_stages.append(stage_summary) print( f"[scan] stage={stage['name']} discovered={len(models)} new_processed={processed_count} " - f"candidates_added={len(stage_candidates)} candidates_total={len(candidates)}/{candidate_goal} " + f"routes_added={len(stage_candidates)} routes_total={len(candidates)} " + f"candidate_models={candidate_model_count(candidates)}/{candidate_goal} " f"skipped_added={len(skipped) - skipped_before_stage}", flush=True, ) @@ -1391,7 +1458,7 @@ def run_submission( uniqueness_rejected_candidates: list[dict[str, Any]] = [] target_submit_count = resolve_max_submit_count( args=args, - planned_count=len(candidates), + planned_count=candidate_model_count(candidates), remaining_daily_quota=remaining_daily_quota, ) if strategy_manager is not None: @@ -1642,10 +1709,13 @@ def run_submission( "olderHistoryScanEnabled": allow_older_models_for_scan, }, "scanLimit": scan_limit, + "rawCandidateGoal": raw_candidate_goal, + "candidatePoolLimit": candidate_pool_limit, "candidateGoal": candidate_goal, "scanStages": scan_stages, "scannedModels": scanned_model_count, "candidateCount": len(candidates), + "candidateModelCount": candidate_model_count(candidates), "candidateFrameworkCounts": dict(candidate_framework_counts.most_common()), "targetSubmitCount": target_submit_count, "maxSubmitAttempts": 0 if args.dry_run else min( diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index f5d92b89..a70ea123 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -3,9 +3,12 @@ from __future__ import annotations from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta +import gzip +import json import os from pathlib import Path from typing import Any +import uuid from architecture_compatibility import ( DEFAULT_ARCHITECTURE_BLOCK_TTL_DAYS, @@ -15,7 +18,7 @@ from architecture_compatibility import ( architecture_profile, architecture_profiles, ) -from common import append_jsonl, parse_datetime, read_jsonl, update_jsonl, utc_now +from common import append_jsonl, parse_datetime, read_json, read_jsonl, update_jsonl, utc_now, write_json, write_jsonl 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 @@ -24,6 +27,12 @@ from task_registry import task_type_from_history_task DEFAULT_OUTCOMES_PATH = Path("outcomes/submissions.jsonl") +DEFAULT_OUTCOME_CHECKPOINT_PATH = Path(".modelhub_state/outcome_checkpoint.json") +DEFAULT_RECENT_OUTCOMES_PATH = Path(".modelhub_state/recent_outcomes.jsonl") +DEFAULT_ARCHIVE_PENDING_DIR = Path(".modelhub_state/archive_pending/outcomes") +OUTCOME_CHECKPOINT_VERSION = 1 +DEFAULT_OUTCOME_COMPACT_THRESHOLD = 2000 +DEFAULT_RECENT_OUTCOME_LIMIT = 1000 FAILURE_ENRICHMENT_LIMIT = 40 FAILURE_ENRICHMENT_WORKERS = 4 FAILURE_ENRICHMENT_MAX_ATTEMPTS = 3 @@ -35,9 +44,21 @@ def _now_iso() -> str: class OutcomeTracker: - def __init__(self, path: Path | str) -> None: + def __init__( + self, + path: Path | str, + *, + checkpoint_path: Path | str = DEFAULT_OUTCOME_CHECKPOINT_PATH, + recent_path: Path | str = DEFAULT_RECENT_OUTCOMES_PATH, + archive_pending_dir: Path | str = DEFAULT_ARCHIVE_PENDING_DIR, + ) -> None: self.path = Path(path) + self.checkpoint_path = Path(checkpoint_path) + self.recent_path = Path(recent_path) + self.archive_pending_dir = Path(archive_pending_dir) self._records: list[dict[str, Any]] = [] + self._recent_records: list[dict[str, Any]] = read_jsonl(self.recent_path) + self._checkpoint: dict[str, Any] = self._load_checkpoint() self._by_task_id: dict[str, dict[str, Any]] = {} self._by_model_gpu: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) self._failed_model_gpus: dict[tuple[str, str], datetime] = {} @@ -46,13 +67,131 @@ class OutcomeTracker: self._records = read_jsonl(self.path) self._rebuild_indexes() + self._compact_if_needed(force=not bool(self._checkpoint) and len(self._records) > self._compact_threshold()) + last_sync_times = [ parse_datetime(record.get("lastSyncTime")) - for record in self._records + for record in [*self._records, *self._recent_records] if record.get("lastSyncTime") ] + checkpoint_sync = parse_datetime(self._checkpoint.get("lastSyncTime")) + if checkpoint_sync is not None: + last_sync_times.append(checkpoint_sync) self._last_sync_time: datetime = max(last_sync_times) if last_sync_times else utc_now() - timedelta(days=7) + def _compact_threshold(self) -> int: + raw = os.getenv("MODELHUB_AGENT_OUTCOME_COMPACT_THRESHOLD", str(DEFAULT_OUTCOME_COMPACT_THRESHOLD)) + try: + return max(500, int(raw)) + except ValueError: + return DEFAULT_OUTCOME_COMPACT_THRESHOLD + + def _recent_limit(self) -> int: + raw = os.getenv("MODELHUB_AGENT_RECENT_OUTCOME_LIMIT", str(DEFAULT_RECENT_OUTCOME_LIMIT)) + try: + return max(100, int(raw)) + except ValueError: + return DEFAULT_RECENT_OUTCOME_LIMIT + + def _load_checkpoint(self) -> dict[str, Any]: + try: + payload = read_json(self.checkpoint_path) + except (FileNotFoundError, ValueError, TypeError): + return {} + if not isinstance(payload, dict) or int(payload.get("version") or 0) != OUTCOME_CHECKPOINT_VERSION: + return {} + return payload + + @property + def has_durable_checkpoint(self) -> bool: + return bool(self._checkpoint and isinstance(self._checkpoint.get("report"), dict)) + + def _combined_recent_terminal(self) -> list[dict[str, Any]]: + by_key: dict[str, dict[str, Any]] = {} + for record in [*self._recent_records, *self._records]: + if record.get("outcome") not in {"success", "failed"}: + continue + key = _outcome_record_key(record) + existing = by_key.get(key) + by_key[key] = record if existing is None else _prefer_newer_outcome(existing, record) + return sorted(by_key.values(), key=_outcome_record_timestamp, reverse=True)[: self._recent_limit()] + + @staticmethod + def _archive_record(record: dict[str, Any]) -> dict[str, Any]: + return { + key: value + for key, value in record.items() + if not any( + marker in key.casefold() + for marker in ("url", "token", "cookie", "authorization", "configparams", "response") + ) + } + + def _write_archive_shard(self, records: list[dict[str, Any]]) -> str | None: + if not records: + return None + now = utc_now() + month = now.strftime("%Y-%m") + name = f"{now.strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:10]}.jsonl.gz" + path = self.archive_pending_dir / month / name + path.parent.mkdir(parents=True, exist_ok=True) + with gzip.open(path, "wt", encoding="utf-8", compresslevel=6) as handle: + for record in records: + handle.write(json.dumps(self._archive_record(record), ensure_ascii=False, sort_keys=True) + "\n") + return f"{month}/{name}" + + def _compact_if_needed(self, *, force: bool = False) -> bool: + if not force and len(self._records) <= self._compact_threshold(): + return False + removed = [ + record + for record in self._records + if record.get("outcome") in {"success", "failed", "policy_cancelled"} + ] + if not removed: + return False + retained = [ + record + for record in self._records + if record.get("outcome") not in {"success", "failed", "policy_cancelled"} + ] + full_report = self.get_stats_report() + recent_records = self._combined_recent_terminal() + shard = self._write_archive_shard(removed) + archived_total = max(0, int(full_report.get("totalRecords") or 0) - len(retained)) + checkpoint_report = dict(full_report) + checkpoint_report.update( + { + "totalRecords": archived_total, + "pendingRecords": 0, + } + ) + previous_shards = list(self._checkpoint.get("archiveShards") or []) + if shard: + previous_shards.append(shard) + sync_times = [ + parse_datetime(record.get("lastSyncTime")) + for record in [*removed, *retained] + if record.get("lastSyncTime") + ] + last_sync = max((value for value in sync_times if value is not None), default=None) + self._checkpoint = { + "version": OUTCOME_CHECKPOINT_VERSION, + "generatedAt": utc_now().isoformat(), + "lastSyncTime": last_sync.isoformat() if last_sync else self._checkpoint.get("lastSyncTime"), + "archivedRecords": archived_total, + "archiveShards": previous_shards[-200:], + "recentLimit": self._recent_limit(), + "report": checkpoint_report, + } + self._records = retained + self._recent_records = recent_records + write_json(self.checkpoint_path, self._checkpoint) + write_jsonl(self.recent_path, self._recent_records) + write_jsonl(self.path, self._records) + self._rebuild_indexes() + return True + def set_failure_llm_classifier(self, classifier: LLMAssistedClassifier | None) -> None: self._failure_llm_classifier = classifier @@ -79,7 +218,9 @@ class OutcomeTracker: def get_strategy_history_records(self) -> list[dict[str, Any]]: """Expose only successes and evidence-attributable failures for GPU ranking.""" records: list[dict[str, Any]] = [] - for record in self._records: + recent_by_key = {_outcome_record_key(record): record for record in self._recent_records} + recent_by_key.update({_outcome_record_key(record): record for record in self._records}) + for record in recent_by_key.values(): outcome = record.get("outcome") if outcome == "success": verify_result = 1 @@ -101,7 +242,13 @@ class OutcomeTracker: def _rebuild_indexes(self) -> None: self._by_task_id.clear() self._by_model_gpu.clear() - for record in self._records: + # Keep the bounded recent window in the identity index as well. The + # platform history API uses an inclusive time cursor, so the first page + # after a restart can contain terminal tasks already represented by the + # checkpoint. Remembering their task IDs prevents double-counting + # without loading the cold archive. + indexed_records = [*self._recent_records, *self._records] + for record in indexed_records: task_id = record.get("taskId") if task_id: self._by_task_id[str(task_id)] = record @@ -487,12 +634,15 @@ class OutcomeTracker: now_datetime = utc_now() now = now_datetime.isoformat() terminal = [r for r in self._records if r.get("outcome") in {"success", "failed"}] + recent_terminal = self._combined_recent_terminal() gpu_groups: dict[str, list[dict[str, Any]]] = defaultdict(list) 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) 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: gpu = record.get("targetGpu") or "unknown" @@ -514,6 +664,17 @@ class OutcomeTracker: size_bucket = int(load_bytes).bit_length() - 1 sized_profile_groups[(gpu, fw, tt, model_type, quantization, size_bucket)].append(record) + for record in recent_terminal: + gpu = record.get("targetGpu") or "unknown" + fw = record.get("framework") or "unknown" + tt = record.get("taskType") or "unknown" + recent_combo_groups[(gpu, fw, tt)].append(record) + profile = record.get("modelProfile") or {} + model_type = str(profile.get("modelType") or "").strip() + if model_type: + quantization = str(profile.get("quantizationMethod") or "none").strip() + recent_profile_groups[(gpu, fw, tt, model_type, quantization)].append(record) + gpu_summaries = {gpu: _summarize(records) for gpu, records in gpu_groups.items()} framework_summaries = {fw: _summarize(records) for fw, records in framework_groups.items()} combination_stats = { @@ -521,7 +682,7 @@ class OutcomeTracker: for (gpu, fw, tt), records in combo_groups.items() } recent_combination_stats: dict[str, dict[str, Any]] = {} - for (gpu, fw, tt), records in combo_groups.items(): + for (gpu, fw, tt), records in recent_combo_groups.items(): recent = sorted(records, key=_outcome_record_timestamp, reverse=True)[:20] consecutive_failures = _consecutive_attributable_failures(recent) consecutive_platform_failures = _consecutive_platform_failures(recent) @@ -554,6 +715,8 @@ class OutcomeTracker: "quantizationMethod": quantization, **_summarize(records), } + for (gpu, fw, tt, model_type, quantization), records in recent_profile_groups.items(): + key = f"{gpu}|{fw}|{tt}|{model_type}|{quantization}" recent = sorted(records, key=_outcome_record_timestamp, reverse=True)[:20] consecutive_failures = _consecutive_attributable_failures(recent) last_terminal_at = None @@ -563,7 +726,11 @@ class OutcomeTracker: or parse_datetime(recent[0].get("submitTime")) ) recent_profile_combination_stats[key] = { - **profile_combination_stats[key], + "targetGpu": gpu, + "framework": fw, + "taskType": tt, + "modelType": model_type, + "quantizationMethod": quantization, **_summarize(recent), "consecutiveFailures": consecutive_failures, "lastTerminalAt": last_terminal_at.isoformat() if last_terminal_at else None, @@ -607,7 +774,7 @@ class OutcomeTracker: architecture_block_ttl_days = _architecture_block_ttl_days() architecture_blocks = _build_architecture_compatibility_blocks( - terminal, + recent_terminal, now=now_datetime, ttl_days=architecture_block_ttl_days, ) @@ -616,7 +783,7 @@ class OutcomeTracker: combination = f"{block['targetGpu']}|{block['framework']}" architecture_blocks_by_gpu_framework[combination] += 1 - return { + report = { "generatedAt": now, "totalRecords": len(self._records), "pendingRecords": pending_count, @@ -639,6 +806,102 @@ class OutcomeTracker: "totals": _summarize(terminal), "warnings": warnings, } + return self._merge_checkpoint_report(report, recent_terminal=recent_terminal, now=now_datetime) + + def _merge_checkpoint_report( + self, + report: dict[str, Any], + *, + recent_terminal: list[dict[str, Any]], + now: datetime, + ) -> dict[str, Any]: + baseline = self._checkpoint.get("report") + if not isinstance(baseline, dict): + return report + merged = dict(report) + for field in ( + "gpuSummaries", + "frameworkSummaries", + "combinationStats", + "profileCombinationStats", + "sizedProfileCombinationStats", + ): + base_items = baseline.get(field) if isinstance(baseline.get(field), dict) else {} + delta_items = report.get(field) if isinstance(report.get(field), dict) else {} + combined: dict[str, dict[str, Any]] = {} + for key in set(base_items) | set(delta_items): + combined[str(key)] = _merge_stat_items(base_items.get(key), delta_items.get(key)) + merged[field] = combined + + merged["totals"] = _merge_stat_items(baseline.get("totals"), report.get("totals")) + merged["totalRecords"] = int(self._checkpoint.get("archivedRecords") or 0) + len(self._records) + merged["terminalRecords"] = int((merged.get("totals") or {}).get("total") or 0) + merged["pendingRecords"] = sum(1 for record in self._records if record.get("outcome") == "pending") + merged["policyCancelledRecords"] = int(baseline.get("policyCancelledRecords") or 0) + sum( + 1 for record in self._records if record.get("outcome") == "policy_cancelled" + ) + + observed: dict[str, float] = {} + for source in (baseline.get("observedGpuMemoryGiB") or {}, report.get("observedGpuMemoryGiB") or {}): + if not isinstance(source, dict): + continue + for gpu, raw in source.items(): + try: + value = float(raw) + except (TypeError, ValueError): + continue + previous = observed.get(str(gpu)) + observed[str(gpu)] = min(previous, value) if previous else value + merged["observedGpuMemoryGiB"] = observed + + blocks = { + str(key): dict(value) + for key, value in (baseline.get("architectureCompatibilityBlocks") or {}).items() + if isinstance(value, dict) + and (parse_datetime(value.get("expiresAt")) or now + timedelta(seconds=1)) > now + } + for record in recent_terminal: + if record.get("outcome") != "success": + continue + event_time = parse_datetime(record.get("lastSyncTime")) or parse_datetime(record.get("submitTime")) + if event_time is None: + continue + for profile in _stored_architecture_profiles(record, include_model_type=True): + key = architecture_compatibility_key( + str(record.get("targetGpu") or ""), + str(record.get("framework") or ""), + str(record.get("taskType") or ""), + profile["signature"], + ) + block = blocks.get(str(key)) if key is not None else None + if block is not None and event_time >= (parse_datetime(block.get("latestFailureAt")) or event_time): + blocks.pop(str(key), None) + blocks.update( + { + str(key): dict(value) + for key, value in (report.get("architectureCompatibilityBlocks") or {}).items() + if isinstance(value, dict) + } + ) + by_gpu_framework: dict[str, int] = defaultdict(int) + for block in blocks.values(): + by_gpu_framework[f"{block.get('targetGpu')}|{block.get('framework')}"] += 1 + merged["architectureCompatibilityBlocks"] = blocks + merged["architectureCompatibilitySummary"] = { + "activeBlockCount": len(blocks), + "ttlDays": _architecture_block_ttl_days(), + "byGpuFramework": dict(by_gpu_framework), + } + + warnings: list[str] = [] + for gpu, summary in (merged.get("gpuSummaries") or {}).items(): + if int(summary.get("decisionTotal") or 0) >= 4 and float(summary.get("decisionFailureRate") or 0) >= 0.5: + warnings.append(f"GPU {gpu} 本地统计失败率偏高（≥50%），建议重点关注。") + for key, stat in (merged.get("combinationStats") or {}).items(): + if int(stat.get("decisionTotal") or 0) >= 3 and float(stat.get("decisionFailureRate") or 0) >= 0.6: + warnings.append(f"组合 {key} 近期失败集中，建议降低该 GPU+框架的提交优先级。") + merged["warnings"] = warnings + return merged def save(self) -> None: local_records = list(self._records) @@ -658,11 +921,14 @@ class OutcomeTracker: self._records = update_jsonl(self.path, merge) self._rebuild_indexes() + self._compact_if_needed() def _rebuild_failed_index(self) -> None: self._failed_model_gpus.clear() latest_by_combo: dict[tuple[str, str], tuple[datetime, dict[str, Any]]] = {} - for record in self._records: + recent_by_key = {_outcome_record_key(record): record for record in self._recent_records} + recent_by_key.update({_outcome_record_key(record): record for record in self._records}) + for record in recent_by_key.values(): # Platform, unresolved, and policy outcomes neither clear nor # create a model/GPU cooldown. The full-history audit showed that # more than half of failures lack enough evidence for attribution. @@ -781,6 +1047,42 @@ def _summarize(records: list[dict[str, Any]]) -> dict[str, Any]: } +def _merge_stat_items(base: Any, delta: Any) -> dict[str, Any]: + base = base if isinstance(base, dict) else {} + delta = delta if isinstance(delta, dict) else {} + merged = {**base, **delta} + count_fields = ( + "total", + "successCount", + "failureCount", + "attributableFailureCount", + "platformFailureCount", + "unresolvedFailureCount", + "decisionTotal", + "pendingCount", + ) + for field in count_fields: + merged[field] = int(base.get(field) or 0) + int(delta.get(field) or 0) + breakdown: dict[str, int] = defaultdict(int) + for source in (base.get("failureBreakdown") or {}, delta.get("failureBreakdown") or {}): + if isinstance(source, dict): + for key, value in source.items(): + breakdown[str(key)] += int(value or 0) + merged["failureBreakdown"] = dict(breakdown) + total = merged["total"] + decisions = merged["decisionTotal"] + merged.update( + { + "successRate": round(merged["successCount"] / total, 4) if total else 0.0, + "failureRate": round(merged["failureCount"] / total, 4) if total else 0.0, + "decisionSuccessRate": round(merged["successCount"] / decisions, 4) if decisions else 0.0, + "decisionFailureRate": round(merged["attributableFailureCount"] / decisions, 4) if decisions else 0.0, + "pendingRate": round(merged["pendingCount"] / total, 4) if total else 0.0, + } + ) + return merged + + def _architecture_block_ttl_days() -> int: raw = os.getenv("MODELHUB_ARCHITECTURE_BLOCK_TTL_DAYS") if raw is None: diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 5b294752..153bed2f 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -572,12 +572,27 @@ def run_poll_loop( architecture_bootstrap_summary: dict[str, Any] = {"enabled": False} if not getattr(base_args, "skip_outcome_sync", False): try: - architecture_bootstrap_summary = _bootstrap_architecture_history( - modelhub_client=modelhub_client, - outcome_tracker=outcome_tracker, - ledger_path=Path(base_args.ledger_path), - now=now, - ) + if outcome_tracker.has_durable_checkpoint: + cached_feedback = outcome_tracker.get_stats_report() + architecture_bootstrap_summary = { + "source": "durable_checkpoint", + "terminalRecords": int(cached_feedback.get("terminalRecords") or 0), + "architectureBlocks": len(cached_feedback.get("architectureCompatibilityBlocks") or {}), + "fullHistoryScanSkipped": True, + } + log( + "[architecture-bootstrap] source=durable_checkpoint " + f"terminal={architecture_bootstrap_summary['terminalRecords']} " + f"blocks={architecture_bootstrap_summary['architectureBlocks']} " + "full_history_scan=skipped" + ) + else: + architecture_bootstrap_summary = _bootstrap_architecture_history( + modelhub_client=modelhub_client, + outcome_tracker=outcome_tracker, + ledger_path=Path(base_args.ledger_path), + now=now, + ) architecture_bootstrap_summary["enabled"] = True _persist_architecture_blacklist( outcome_tracker.get_stats_report(), diff --git a/modelhub_submmit_api/state_sync.py b/modelhub_submmit_api/state_sync.py index ac276d77..d2545444 100644 --- a/modelhub_submmit_api/state_sync.py +++ b/modelhub_submmit_api/state_sync.py @@ -25,6 +25,7 @@ from version import AGENT_VERSION STATE_SCHEMA_VERSION = 1 DEFAULT_REMOTE = "https://dev.modelhub.org.cn/CoolBoy/submmit.git" DEFAULT_BRANCH = "agent-state" +DEFAULT_ARCHIVE_BRANCH_PREFIX = "agent-archive" DEFAULT_BATCH_SIZE = 20 DEFAULT_RETENTION_DAYS = 30 DEFAULT_HISTORY_DEPTH = 20 @@ -37,7 +38,9 @@ STATE_ALLOWLIST = ( ".modelhub_state/gpu_strategy.json", ".modelhub_state/market_intelligence.json", ".modelhub_state/official_capabilities.json", + ".modelhub_state/outcome_checkpoint.json", ".modelhub_state/queue_cleanup_latest.json", + ".modelhub_state/recent_outcomes.jsonl", ".modelhub_state/recovery_active_tasks.jsonl", ".modelhub_state/recovery_intents.jsonl", ".modelhub_state/routing_intelligence.json", @@ -205,10 +208,103 @@ class StateGitSync: return f"{name} <{email}>".encode("utf-8") def _remote_oid(self) -> str | None: + return self._remote_oid_for(self.branch) + + def _remote_oid_for(self, branch: str) -> str | None: result = porcelain.ls_remote(self.remote, quiet=True, **self._auth_kwargs()) - oid = result.refs.get(f"refs/heads/{self.branch}".encode("utf-8")) + oid = result.refs.get(f"refs/heads/{branch}".encode("utf-8")) return oid.decode("ascii") if oid else None + def _sync_archive_pending(self) -> None: + pending_root = self.state_dir / "archive_pending" / "outcomes" + files = sorted(path for path in pending_root.glob("*/*.jsonl.gz") if path.is_file()) + if not files: + return + by_month: dict[str, list[Path]] = {} + for path in files: + by_month.setdefault(path.parent.name, []).append(path) + for month, month_files in sorted(by_month.items()): + branch = f"{DEFAULT_ARCHIVE_BRANCH_PREFIX}-{month}" + remote_oid = self._remote_oid_for(branch) + parent = Path(tempfile.mkdtemp(prefix="modelhub-agent-archive-")) + workspace = parent / "archive" + try: + if remote_oid: + porcelain.clone( + self.remote, + workspace, + branch=branch, + depth=1, + checkout=True, + errstream=io.BytesIO(), + **self._auth_kwargs(), + ) + else: + workspace.mkdir(parents=True) + repo = porcelain.init(workspace) + repo.refs.set_symbolic_ref(b"HEAD", f"refs/heads/{branch}".encode("utf-8")) + repo = Repo(str(workspace)) + manifest_path = workspace / "manifest.json" + try: + manifest = read_json(manifest_path) + except (FileNotFoundError, ValueError, TypeError): + manifest = {} + archived = manifest.get("files") if isinstance(manifest.get("files"), dict) else {} + for source in month_files: + destination = workspace / "outcomes" / source.name + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + archived[f"outcomes/{source.name}"] = { + "sha256": _sha256_file(destination), + "bytes": destination.stat().st_size, + } + write_json( + manifest_path, + { + "schemaVersion": 1, + "month": month, + "updatedAt": _utc_now().isoformat(), + "files": archived, + }, + ) + manifest_path.with_name(f".{manifest_path.name}.lock").unlink(missing_ok=True) + porcelain.add(repo) + status = porcelain.status(repo) + if any(status.staged.get(kind) for kind in ("add", "delete", "modify")): + porcelain.commit( + repo, + message=f"archive: outcomes {month}".encode("utf-8"), + author=self._author, + committer=self._author, + ) + if self._remote_oid_for(branch) != remote_oid: + raise StateSyncError(f"archive branch {branch} changed remotely") + porcelain.push( + repo, + self.remote, + refspecs=f"HEAD:refs/heads/{branch}", + force=True, + outstream=io.BytesIO(), + errstream=io.BytesIO(), + **self._auth_kwargs(), + ) + if self._remote_oid_for(branch) != repo.head().decode("ascii"): + raise StateSyncError(f"archive branch {branch} verification failed") + for source in month_files: + source.unlink(missing_ok=True) + self.log( + f"[archive-sync] branch={branch} shards={len(month_files)} " + f"files_total={len(archived)} status=ok" + ) + finally: + shutil.rmtree(parent, ignore_errors=True) + + def _sync_archive_pending_safely(self) -> None: + try: + self._sync_archive_pending() + except Exception as exc: + self.log(f"[archive-sync] status=deferred reason={_safe_text(exc)}") + def _create_workspace(self, remote_oid: str | None) -> None: parent = Path(tempfile.mkdtemp(prefix="modelhub-agent-state-")) workspace = parent / "state" @@ -539,7 +635,7 @@ class StateGitSync: # A restored snapshot must refresh official configs before routing. sanitized_market["frameworkUpdatedAt"] = None write_json(destination, sanitized_market) - elif relative == "outcomes/submissions.jsonl": + elif relative in {"outcomes/submissions.jsonl", ".modelhub_state/recent_outcomes.jsonl"}: sanitized_outcomes: list[dict[str, Any]] = [] for row in read_jsonl(source): sanitized_outcomes.append( @@ -593,6 +689,10 @@ class StateGitSync: staged = any(status.staged.get(kind) for kind in ("add", "delete", "modify")) if not staged: self.healthy = True + # A previous archive push may have been deferred while the + # hot snapshot was already current. Retry cold shards even + # when this cycle has no hot-state commit to publish. + self._sync_archive_pending_safely() return True porcelain.commit( repo, @@ -631,6 +731,7 @@ class StateGitSync: self.last_error = None self.healthy = True self.log(f"[state-sync] generation={self.generation} phase={phase} status=ok") + self._sync_archive_pending_safely() return True except Exception as exc: self.healthy = False diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 86a18f17..6219aa10 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.21.1" +AGENT_VERSION = "2026.08.22.1" diff --git a/tests/test_modelscope_discovery.py b/tests/test_modelscope_discovery.py index c6ee8ce8..e174a352 100644 --- a/tests/test_modelscope_discovery.py +++ b/tests/test_modelscope_discovery.py @@ -3,6 +3,7 @@ from __future__ import annotations import sys import unittest from pathlib import Path +from unittest.mock import patch PACKAGE_DIR = Path(__file__).resolve().parents[1] / "modelhub_submmit_api" @@ -68,6 +69,17 @@ class ModelScopeDiscoveryTests(unittest.TestCase): self.assertEqual(1786294389, int(first.timestamp())) # type: ignore[union-attr] self.assertEqual(1, metadata_client.calls) + def test_model_detail_cache_evicts_old_entries_at_fixed_limit(self) -> None: + metadata_client = ModelMetadataClient() + with patch.dict("os.environ", {"MODELSCOPE_DETAIL_CACHE_MAX_MODELS": "32"}): + discovery = HuggingFaceDiscovery(legacy_http_client=metadata_client) # type: ignore[arg-type] + for index in range(40): + discovery.get_model_last_modified(f"owner/model-{index}") + + self.assertEqual(32, len(discovery._model_last_modified_cache)) + discovery.get_model_last_modified("owner/model-0") + self.assertEqual(41, metadata_client.calls) + def test_openapi_page_size_never_exceeds_platform_limit(self) -> None: http_client = RecordingHttpClient() discovery = HuggingFaceDiscovery(http_client=http_client) # type: ignore[arg-type] diff --git a/tests/test_super_agent.py b/tests/test_super_agent.py index 9e9d2f4a..4b3d6a7a 100644 --- a/tests/test_super_agent.py +++ b/tests/test_super_agent.py @@ -1,5 +1,7 @@ from __future__ import annotations +import os +import gzip import tempfile import unittest from datetime import datetime, timezone @@ -19,6 +21,7 @@ from common import read_jsonl, write_json, write_jsonl # noqa: E402 from config_optimizer import SafeConfigOptimizer # noqa: E402 from hf_discovery import HuggingFaceDiscovery, parse_model_card_front_matter # noqa: E402 from official_capabilities import OfficialCapabilityRegistry # noqa: E402 +from outcome_tracker import OutcomeTracker # noqa: E402 from routing_engine import SuccessFirstRoutingEngine # noqa: E402 from state_sync import StateGitSync # noqa: E402 @@ -44,6 +47,91 @@ class OfficialClient: class SuperAgentTests(unittest.TestCase): + def test_outcome_history_compacts_to_checkpoint_recent_window_and_gzip_archive(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + outcomes = root / "outcomes.jsonl" + checkpoint = root / "checkpoint.json" + recent = root / "recent.jsonl" + archive = root / "archive" + rows = [ + { + "taskId": str(index), + "modelId": f"owner/model-{index}", + "targetGpu": "gpu-a", + "framework": "vllm", + "taskType": "text-generation", + "submitTime": f"2026-08-{1 + index // 100:02d}T00:00:{index % 60:02d}+00:00", + "lastSyncTime": "2026-08-21T00:00:00+00:00", + "outcome": "success" if index % 2 == 0 else "failed", + "failureCategory": "model_runtime" if index % 2 else None, + "failureScope": "model" if index % 2 else None, + "logCosUrl": "https://secret.invalid/signed?token=hidden", + } + for index in range(600) + ] + write_jsonl(outcomes, rows) + with patch.dict( + os.environ, + { + "MODELHUB_AGENT_OUTCOME_COMPACT_THRESHOLD": "500", + "MODELHUB_AGENT_RECENT_OUTCOME_LIMIT": "100", + }, + ): + tracker = OutcomeTracker( + outcomes, + checkpoint_path=checkpoint, + recent_path=recent, + archive_pending_dir=archive, + ) + self.assertTrue(tracker.has_durable_checkpoint) + self.assertEqual([], read_jsonl(outcomes)) + self.assertEqual(100, len(read_jsonl(recent))) + report = tracker.get_stats_report() + self.assertEqual(600, report["terminalRecords"]) + self.assertEqual(300, report["totals"]["successCount"]) + shard = next(archive.rglob("*.jsonl.gz")) + import gzip + + with gzip.open(shard, "rt", encoding="utf-8") as handle: + archived_text = handle.read() + self.assertNotIn("logCosUrl", archived_text) + self.assertNotIn("token=hidden", archived_text) + + restored = OutcomeTracker( + outcomes, + checkpoint_path=checkpoint, + recent_path=recent, + archive_pending_dir=archive, + ) + self.assertEqual(600, restored.get_stats_report()["terminalRecords"]) + + restored.record_submission( + model_id="owner/new-model", + target_gpu="gpu-a", + framework="vllm", + task_type="text-generation", + task_id="new-task", + submit_time="2026-08-22T00:00:00+00:00", + ) + restored._update_record_from_task( + restored._by_task_id["new-task"], + {"status": "success", "verifyResult": 1}, + ) + restored.save() + updated = restored.get_stats_report() + self.assertEqual(601, updated["terminalRecords"]) + self.assertEqual(301, updated["totals"]["successCount"]) + + restarted = OutcomeTracker( + outcomes, + checkpoint_path=checkpoint, + recent_path=recent, + archive_pending_dir=archive, + ) + self.assertEqual(601, restarted.get_stats_report()["terminalRecords"]) + self.assertIn("599", restarted._by_task_id) + def test_official_registry_discovers_catalog_and_exact_model_routes(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: registry = OfficialCapabilityRegistry(Path(temporary_dir) / "official.json", log_fn=lambda _: None) @@ -149,6 +237,17 @@ class SuperAgentTests(unittest.TestCase): restored_project / ".modelhub_state" / "worker_crashes.jsonl", [{"at": "2026-08-21T01:00:00+00:00", "exitCode": 1}], ) + pending_archive = ( + project + / ".modelhub_state" + / "archive_pending" + / "outcomes" + / "2026-08" + / "shard.jsonl.gz" + ) + pending_archive.parent.mkdir(parents=True, exist_ok=True) + with gzip.open(pending_archive, "wt", encoding="utf-8") as handle: + handle.write('{"taskId":"archived"}\n') credentials = {"username": "tester", "email": "tester@example.com", "password": "secret-value"} manager = StateGitSync( project_root=project, @@ -171,6 +270,9 @@ class SuperAgentTests(unittest.TestCase): ] ) self.assertIsNotNone(batch_id) + self.assertFalse(pending_archive.exists()) + archive_refs = porcelain.ls_remote(str(remote)).refs + self.assertIn(b"refs/heads/agent-archive-2026-08", archive_refs) manager.close() restored = StateGitSync(