From d908706f9a5f5b8bdb41f1ab0742dcef64bf112f Mon Sep 17 00:00:00 2001 From: CoolBoy Date: Wed, 12 Aug 2026 01:04:34 +0800 Subject: [PATCH] fix: protect running validation tasks from queue cleanup --- README.md | 16 +++-- modelhub_submmit_api/README.md | 11 ++- modelhub_submmit_api/outcome_tracker.py | 72 +++++++++++++++++-- modelhub_submmit_api/poll_runner.py | 7 ++ modelhub_submmit_api/queue_cleanup.py | 89 +++++++++++++++++++++-- modelhub_submmit_api/version.py | 2 +- tests/test_candidate_preflight.py | 72 +++++++++++++++++++ tests/test_queue_cleanup.py | 94 +++++++++++++++++++++++++ 8 files changed, 343 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index eb369727..a456e5e2 100644 --- a/README.md +++ b/README.md @@ -153,9 +153,12 @@ older-than-seven-days tasks beyond that account's current limit minus 10. Later scheduled cleanup uses limit minus 5, retaining a small hysteresis buffer that avoids repeatedly over-pruning valid work. For a 100-task account the two boundaries are 90 and 95; for a 500-task account they are 490 and 495. Recent -overflow tasks are always kept. ModelScope metadata failures fail closed and -never trigger cancellation. Immediately before mutation, task ownership, active -status, and post-OOM queue position are checked again. +overflow tasks are always kept. Age-based cleanup applies only to tasks that are +still waiting; running validation tasks are protected even beyond the boundary. +ModelScope metadata failures fail closed and never trigger cancellation. +Immediately before the age-only stop batch, task ownership, active status, and +post-OOM queue position are checked again. Deterministic OOM cleanup may still +stop a running task because it cannot fit the selected GPU. The verified capacities, safe repository-size boundaries, evidence hierarchy, and source links are recorded in @@ -247,12 +250,15 @@ after OOM cleanup and a second queue check. Version `2026.08.11.4` replaces fixed queue positions with per-account dynamic boundaries derived from each discovered capacity: limit minus 10 for admission and startup cleanup, then limit minus 5 for scheduled dynamic cleanup. +Version `2026.08.12.1` protects running tasks from age-based cleanup and records +worker-initiated stops as `policy_cancelled`, excluding them from success-rate, +failure-cooldown, and circuit-breaker evidence. ## Deploy Create a tag and submit the repository URL plus tag in "我的适配智能体". ```bash -git tag agent-v18 -git push origin agent-v18 +git tag agent-v19 +git push origin agent-v19 ``` diff --git a/modelhub_submmit_api/README.md b/modelhub_submmit_api/README.md index 3aa075cf..28cb3be4 100644 --- a/modelhub_submmit_api/README.md +++ b/modelhub_submmit_api/README.md @@ -124,8 +124,10 @@ bash run_poll.sh --dry-run positions. The account pool enforces this per-account boundary atomically and updates it when capacity probing discovers a higher limit. Cleanup stops OOM tasks first, recalculates the surviving queue order, and applies the same limit-minus-10 boundary on startup. Scheduled - cleanup relaxes to limit minus 5 to avoid excessive pruning. Recent overflow tasks stay, and - unknown ModelScope timestamps never authorize a cancellation. + cleanup relaxes to limit minus 5 to avoid excessive pruning. Age cleanup stops waiting tasks + only; running tasks are protected and rechecked after OOM cleanup, immediately before the + age-only stop batch. Recent + overflow tasks stay, and unknown ModelScope timestamps never authorize a cancellation. ## Important Flags @@ -167,6 +169,11 @@ minus 5, while admission continues to reserve the final 10 slots for recent models. Override these suffix sizes with `MODELHUB_RECENT_MODEL_RESERVE_SLOTS` and `MODELHUB_DYNAMIC_OLD_MODEL_CLEANUP_RESERVE_SLOTS`. +Every successful worker-initiated stop is persisted as `policy_cancelled` in the +outcome store. It is excluded from GPU/framework success rates, local failure +cooldowns, and circuit breakers. If a task races to a real success before the +stop takes effect, that success remains authoritative. + Failure-informed preflight is enabled by default. It rejects deterministic missing-file and predicted-OOM cases, clamps unsafe context-length arguments, and records its decisions in `candidatePreflight` and each candidate's diff --git a/modelhub_submmit_api/outcome_tracker.py b/modelhub_submmit_api/outcome_tracker.py index 34a759b6..3cbac939 100644 --- a/modelhub_submmit_api/outcome_tracker.py +++ b/modelhub_submmit_api/outcome_tracker.py @@ -88,6 +88,49 @@ class OutcomeTracker: self._by_model_gpu[(model_id, target_gpu)].append(record) append_jsonl(self.path, record) + def mark_policy_cancellations(self, decisions: list[dict[str, Any]]) -> int: + """Persist intentional task stops so they never become failure evidence.""" + marked = 0 + marked_at = _now_iso() + for decision in decisions: + task_id_value = decision.get("taskId") + if task_id_value is None: + continue + task_id = str(task_id_value) + existing = self._by_task_id.get(task_id) + if existing is not None and existing.get("outcome") in {"success", "failed"}: + continue + reasons = list(decision.get("cleanupReasons") or [decision.get("reason")]) + reasons = [str(reason) for reason in reasons if reason] + if existing is None: + existing = { + "modelId": decision.get("modelId") or "", + "targetGpu": decision.get("gpuType") or decision.get("targetGpu") or "", + "framework": decision.get("framework") or "", + "taskType": decision.get("taskType") or "", + "taskId": task_id, + "submitTime": decision.get("submitTime") or marked_at, + "verifyResult": None, + } + self._records.append(existing) + self._by_task_id[task_id] = existing + self._by_model_gpu[(existing["modelId"], existing["targetGpu"])].append(existing) + existing.update( + { + "lastSyncTime": marked_at, + "status": "cancellation_requested", + "outcome": "policy_cancelled", + "failReason": None, + "policyCancelled": True, + "policyCancellationReasons": reasons, + "policyCancelledAt": marked_at, + } + ) + marked += 1 + if marked: + self._rebuild_failed_index() + return marked + def sync_from_api(self, client: ModelHubClient | ModelHubClientPool) -> int: try: begin = self._last_sync_time @@ -109,7 +152,7 @@ class OutcomeTracker: existing = self._by_task_id.get(task_id) if existing is not None: - if existing.get("outcome") == "pending": + if existing.get("outcome") in {"pending", "policy_cancelled"}: self._update_record_from_task(existing, task) if existing.get("outcome") == "failed" and existing.get("logCosUrl"): enrichment_candidates.append(existing) @@ -287,6 +330,9 @@ class OutcomeTracker: warnings.append(f"组合 {key} 近期失败集中，建议降低该 GPU+框架的提交优先级。") pending_count = sum(1 for r in self._records if r.get("outcome") == "pending") + policy_cancelled_count = sum( + 1 for record in self._records if record.get("outcome") == "policy_cancelled" + ) observed_gpu_memory: dict[str, float] = {} for record in self._records: gpu = str(record.get("targetGpu") or "") @@ -302,6 +348,7 @@ class OutcomeTracker: "generatedAt": now, "totalRecords": len(self._records), "pendingRecords": pending_count, + "policyCancelledRecords": policy_cancelled_count, "terminalRecords": len(terminal), "gpuSummaries": gpu_summaries, "frameworkSummaries": framework_summaries, @@ -337,9 +384,10 @@ class OutcomeTracker: self._failed_model_gpus.clear() latest_by_combo: dict[tuple[str, str], tuple[datetime, dict[str, Any]]] = {} for record in self._records: - # Infrastructure failures neither clear nor create a model/GPU - # cooldown. Look through them to the latest attributable outcome. - if _is_platform_failure(record): + # Infrastructure failures and our own policy cancellations neither + # clear nor create a model/GPU cooldown. Look through them to the + # latest attributable outcome. + if _is_platform_failure(record) or _is_policy_cancelled(record): continue model_id = record.get("modelId") or "" target_gpu = record.get("targetGpu") or "" @@ -362,9 +410,19 @@ class OutcomeTracker: record["lastSyncTime"] = _now_iso() if task.get("logCosUrl"): record["logCosUrl"] = task.get("logCosUrl") + status = str(task.get("status") or "").strip().lower() if is_success(task): record["outcome"] = "success" record["failReason"] = None + if record.get("policyCancelled"): + record["policyCancellationResolvedAsSuccess"] = True + record["policyCancelled"] = False + elif record.get("policyCancelled"): + # A successful stop may be observed as waiting/running briefly before + # the platform publishes its terminal cancellation state. None of + # those intermediate states should become failure evidence. + record["outcome"] = "policy_cancelled" + record["failReason"] = None elif is_failure(task): record["outcome"] = "failed" record["failReason"] = classify_failure(task) @@ -439,6 +497,10 @@ def _is_platform_failure(record: dict[str, Any]) -> bool: return scope == "platform" or category.startswith("platform_") +def _is_policy_cancelled(record: dict[str, Any]) -> bool: + return bool(record.get("policyCancelled")) or record.get("outcome") == "policy_cancelled" + + def _consecutive_attributable_failures(records: list[dict[str, Any]]) -> int: count = 0 for record in records: @@ -486,7 +548,7 @@ def _outcome_record_key(record: dict[str, Any]) -> str: def _outcome_version(record: dict[str, Any]) -> tuple[int, float, float]: last_sync = parse_datetime(record.get("lastSyncTime")) submit_time = parse_datetime(record.get("submitTime")) - outcome_rank = 1 if record.get("outcome") in {"success", "failed"} else 0 + outcome_rank = 1 if record.get("outcome") in {"success", "failed", "policy_cancelled"} else 0 return ( outcome_rank, last_sync.timestamp() if last_sync else 0.0, diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 59881cfd..43d28791 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -324,6 +324,12 @@ def run_poll_loop( age_reserved_slots=age_cleanup_reserve_slots, log=log, ) + policy_cancelled_recorded = outcome_tracker.mark_policy_cancellations( + cleanup_summary["cancelledTasks"] + ) + if policy_cancelled_recorded: + outcome_tracker.save() + cleanup_summary["policyCancelledRecorded"] = policy_cancelled_recorded write_json( Path( getattr( @@ -344,6 +350,7 @@ def run_poll_loop( "certainOomCount": cleanup_summary["certainOomCount"], "oldOverflowCount": cleanup_summary["oldOverflowCount"], "cancelledCount": cleanup_summary["cancelledCount"], + "policyCancelledRecorded": policy_cancelled_recorded, "stopErrorCount": len(cleanup_summary["stopErrors"]), } ) diff --git a/modelhub_submmit_api/queue_cleanup.py b/modelhub_submmit_api/queue_cleanup.py index b5777449..efe027e2 100644 --- a/modelhub_submmit_api/queue_cleanup.py +++ b/modelhub_submmit_api/queue_cleanup.py @@ -261,6 +261,7 @@ def find_old_overflow_tasks( "recentOverflowTasks": 0, "modelAgeUnknown": 0, "accountThresholdUnknown": 0, + "runningOverflowProtected": 0, } for account_index, account_tasks in sorted(grouped.items()): if account_index in incomplete_accounts: @@ -278,6 +279,9 @@ def find_old_overflow_tasks( for position, task in enumerate(ordered, start=1): if position <= threshold: continue + if task.status != "waiting": + skipped["runningOverflowProtected"] += 1 + continue last_modified = model_last_modified.get(task.model_id) if last_modified is None: skipped["modelAgeUnknown"] += 1 @@ -390,6 +394,7 @@ def cleanup_certain_oom_tasks( (item for item in age_rank_tasks if item.account_index == account_index), key=lambda item: item.task_id, )[queue_thresholds.get(account_index, len(age_rank_tasks)):] + if task.status == "waiting" } model_last_modified, age_errors = _load_model_last_modified( overflow_model_ids, @@ -410,7 +415,8 @@ def cleanup_certain_oom_tasks( f"thresholds={','.join(str(queue_thresholds[index]) for index in sorted(queue_thresholds))} " f"reserve_recent_slots={reserved_slots} recent_days={recent_model_days} " f"recent_overflow={age_skipped['recentOverflowTasks']} " - f"age_unknown={age_skipped['modelAgeUnknown']}" + f"age_unknown={age_skipped['modelAgeUnknown']} " + f"running_protected={age_skipped['runningOverflowProtected']}" ) decisions_by_key: dict[tuple[int, int], dict[str, Any]] = {} @@ -444,8 +450,10 @@ def cleanup_certain_oom_tasks( refreshed_tasks, refresh_errors = collect_active_tasks(clients, read_concurrency=read_concurrency) active_ids_by_account: dict[int, set[int]] = {} active_positions_by_account: dict[int, dict[int, int]] = {} + active_status_by_account: dict[int, dict[int, str]] = {} for task in refreshed_tasks: active_ids_by_account.setdefault(task.account_index, set()).add(task.task_id) + active_status_by_account.setdefault(task.account_index, {})[task.task_id] = task.status for account_index in range(len(clients)): planned_oom_ids = { int(decision["taskId"]) @@ -481,14 +489,25 @@ def cleanup_certain_oom_tasks( cleanup_reasons = set(decision.get("cleanupReasons") or [decision.get("reason")]) age_only = cleanup_reasons == {"old_model_beyond_account_queue_threshold"} current_position = active_positions_by_account.get(account_index, {}).get(int(decision["taskId"])) + current_status = active_status_by_account.get(account_index, {}).get(int(decision["taskId"])) account_queue_threshold = queue_thresholds.get(account_index) if age_only and ( current_position is None or account_queue_threshold is None or current_position <= account_queue_threshold + or current_status != "waiting" ): policy_no_longer_applies.append( - {**decision, "recheckedQueuePosition": current_position} + { + **decision, + "recheckedQueuePosition": current_position, + "recheckedStatus": current_status, + "policyChangeReason": ( + "task_started_running" + if current_status == "running" + else "queue_position_or_status_changed" + ), + } ) continue if current_position is not None: @@ -501,16 +520,72 @@ def cleanup_certain_oom_tasks( if stop_failed: break decisions_by_id = {int(item["taskId"]): item for item in by_account[account_index]} - phase_ids = [ - sorted( + for is_oom_phase in (True, False): + task_ids = sorted( task_id for task_id, decision in decisions_by_id.items() if ("certain_oom_repository_size_exceeds_gpu_capacity" in decision["cleanupReasons"]) == is_oom_phase ) - for is_oom_phase in (True, False) - ] - for task_ids in phase_ids: + if not is_oom_phase and task_ids: + # OOM stops can change actual positions, and a waiting task + # can start running after the account-wide recheck above. + # Re-read this account immediately before its age-only stop. + try: + phase_tasks: dict[int, OwnedTask] = {} + for status in ACTIVE_FILTER_STATUSES: + for task in _fetch_status_tasks( + clients[account_index], + account_index=account_index, + status=status, + ): + phase_tasks[task.task_id] = task + except Exception as exc: + stop_errors.append( + { + "accountIndex": account_index + 1, + "taskIds": task_ids, + "error": f"age_policy_final_recheck_failed: {type(exc).__name__}: {exc}", + } + ) + stop_failed = True + break + + phase_positions = { + task_id: position + for position, task_id in enumerate(sorted(phase_tasks), start=1) + } + eligible_task_ids: list[int] = [] + account_queue_threshold = queue_thresholds.get(account_index) + for task_id in task_ids: + current_task = phase_tasks.get(task_id) + if current_task is None: + disappeared.append(decisions_by_id[task_id]) + continue + current_position = phase_positions.get(task_id) + if ( + account_queue_threshold is None + or current_position is None + or current_position <= account_queue_threshold + or current_task.status != "waiting" + ): + policy_no_longer_applies.append( + { + **decisions_by_id[task_id], + "recheckedQueuePosition": current_position, + "recheckedStatus": current_task.status, + "policyChangeReason": ( + "task_started_running" + if current_task.status == "running" + else "queue_position_or_status_changed" + ), + } + ) + continue + decisions_by_id[task_id]["recheckedQueuePosition"] = current_position + eligible_task_ids.append(task_id) + task_ids = eligible_task_ids + for batch in _chunks(task_ids, batch_size): try: clients[account_index].stop_tasks(batch) diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 33f13eed..89251eb8 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.08.11.4" +AGENT_VERSION = "2026.08.12.1" diff --git a/tests/test_candidate_preflight.py b/tests/test_candidate_preflight.py index fc340a0f..9452a54c 100644 --- a/tests/test_candidate_preflight.py +++ b/tests/test_candidate_preflight.py @@ -391,6 +391,78 @@ class CandidatePreflightTests(unittest.TestCase): self.assertEqual(1, report["profileCombinationStats"][key]["failureCount"]) self.assertEqual(1, report["recentProfileCombinationStats"][key]["consecutiveFailures"]) + def test_policy_cancellation_is_excluded_from_failure_feedback(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + path = Path(temporary_dir) / "outcomes.jsonl" + tracker = OutcomeTracker(path) + tracker.record_submission( + "owner/model", + "gpu", + "vllm", + "text-generation", + "task-policy", + datetime.now(timezone.utc).isoformat(), + ) + marked = tracker.mark_policy_cancellations( + [ + { + "taskId": "task-policy", + "modelId": "owner/model", + "gpuType": "gpu", + "cleanupReasons": ["old_model_beyond_account_queue_threshold"], + } + ] + ) + tracker.save() + tracker.sync_from_api( + TaskClient( + [ + { + "taskId": "task-policy", + "status": "cancelled", + "verifyResult": None, + } + ] + ) # type: ignore[arg-type] + ) + report = tracker.get_stats_report() + + self.assertEqual(1, marked) + self.assertEqual(1, report["policyCancelledRecords"]) + self.assertEqual(0, report["terminalRecords"]) + self.assertEqual({}, report["combinationStats"]) + self.assertFalse(tracker.is_model_gpu_failed("owner/model", "gpu")) + + def test_policy_cancellation_does_not_hide_a_racing_success(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + tracker = OutcomeTracker(Path(temporary_dir) / "outcomes.jsonl") + tracker.record_submission( + "owner/model", + "gpu", + "vllm", + "text-generation", + "task-success", + datetime.now(timezone.utc).isoformat(), + ) + tracker.mark_policy_cancellations( + [{"taskId": "task-success", "modelId": "owner/model", "gpuType": "gpu"}] + ) + tracker.sync_from_api( + TaskClient( + [ + { + "taskId": "task-success", + "status": "success", + "verifyResult": 1, + } + ] + ) # type: ignore[arg-type] + ) + report = tracker.get_stats_report() + + self.assertEqual(0, report["policyCancelledRecords"]) + self.assertEqual(1, report["totals"]["successCount"]) + def test_failure_taxonomy_separates_platform_faults_from_model_faults(self) -> None: platform = classify_failure_report( "EXECUTE_EMPTY_RESULT", diff --git a/tests/test_queue_cleanup.py b/tests/test_queue_cleanup.py index c3ca52b8..73565ec8 100644 --- a/tests/test_queue_cleanup.py +++ b/tests/test_queue_cleanup.py @@ -31,16 +31,22 @@ class FakeQueueClient: *, disappear_on_recheck: bool = False, drop_first_on_recheck: bool = False, + promote_on_waiting_read: int | None = None, ) -> None: self.records = list(records) self.disappear_on_recheck = disappear_on_recheck self.drop_first_on_recheck = drop_first_on_recheck + self.promote_on_waiting_read = promote_on_waiting_read self.waiting_reads = 0 self.stopped: list[list[int]] = [] def list_tasks_page(self, *, status: str, **_kwargs: Any) -> dict[str, Any]: if status == "waiting": self.waiting_reads += 1 + if self.promote_on_waiting_read == self.waiting_reads: + waiting = [record for record in self.records if record["status"] == "waiting"] + if waiting: + max(waiting, key=lambda item: int(item["taskId"]))["status"] = "running" if self.disappear_on_recheck and self.waiting_reads >= 2: records: list[dict[str, Any]] = [] else: @@ -108,6 +114,7 @@ class QueueCleanupTests(unittest.TestCase): for index in range(1, 192) ) tasks.append(OwnedTask(0, 92, "owner/recent", "Iluvatar_bi-100", "waiting")) + tasks.append(OwnedTask(0, 93, "owner/old", "Iluvatar_bi-100", "running")) tasks.append(OwnedTask(1, 1192, "owner/recent", "Iluvatar_bi-100", "waiting")) selected, skipped = find_old_overflow_tasks( @@ -124,6 +131,93 @@ class QueueCleanupTests(unittest.TestCase): self.assertEqual([91, 1191], [item["taskId"] for item in selected]) self.assertEqual([91, 191], [item["queuePosition"] for item in selected]) self.assertEqual(2, skipped["recentOverflowTasks"]) + self.assertEqual(1, skipped["runningOverflowProtected"]) + + def test_age_cleanup_never_stops_running_overflow_task(self) -> None: + records = [ + { + "taskId": index, + "modelId": "owner/old", + "gpuType": "Iluvatar_bi-100", + "status": "running" if index == 91 else "waiting", + } + for index in range(1, 92) + ] + client = FakeQueueClient(records) + pool = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + recent_model_reserve_slots=10, + ) + summary = cleanup_certain_oom_tasks( + pool, + FakeDiscovery( + {"owner/old": 1 * GIB}, + {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, + ), # type: ignore[arg-type] + reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), + log=lambda _message: None, + ) + + self.assertEqual(0, summary["oldOverflowCount"]) + self.assertEqual(1, summary["agePolicySkipped"]["runningOverflowProtected"]) + self.assertEqual([], client.stopped) + + def test_age_cleanup_recheck_releases_task_that_started_running(self) -> None: + records = [ + { + "taskId": index, + "modelId": "owner/old", + "gpuType": "Iluvatar_bi-100", + "status": "waiting", + } + for index in range(1, 92) + ] + # Read 1 is discovery, read 2 is the account-wide mutation recheck, + # and read 3 is the final age-only recheck after the OOM phase. + client = FakeQueueClient(records, promote_on_waiting_read=3) + pool = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + recent_model_reserve_slots=10, + ) + summary = cleanup_certain_oom_tasks( + pool, + FakeDiscovery( + {"owner/old": 1 * GIB}, + {"owner/old": datetime(2026, 7, 1, tzinfo=timezone.utc)}, + ), # type: ignore[arg-type] + read_concurrency=1, + reference_time=datetime(2026, 8, 11, tzinfo=timezone.utc), + log=lambda _message: None, + ) + + self.assertEqual(1, summary["oldOverflowCount"]) + self.assertEqual(0, summary["cancelledCount"]) + self.assertEqual("task_started_running", summary["policyNoLongerAppliesTasks"][0]["policyChangeReason"]) + self.assertEqual([], client.stopped) + + def test_certain_oom_cleanup_can_still_stop_running_task(self) -> None: + client = FakeQueueClient( + [ + { + "taskId": 1, + "modelId": "owner/large", + "gpuType": "Iluvatar_bi-100", + "status": "running", + } + ] + ) + pool = ModelHubClientPool([client], active_task_cap=100) # type: ignore[list-item] + summary = cleanup_certain_oom_tasks( + pool, + FakeDiscovery({"owner/large": 40 * GIB}), # type: ignore[arg-type] + log=lambda _message: None, + ) + + self.assertEqual(1, summary["certainOomCount"]) + self.assertEqual(1, summary["cancelledCount"]) + self.assertEqual([[1]], client.stopped) def test_old_overflow_task_is_not_stopped_if_it_moves_inside_dynamic_threshold(self) -> None: records = [