From 9f265595e739147da285887fc05ccc876d2a4799 Mon Sep 17 00:00:00 2001 From: Victor Cao Date: Sat, 17 Oct 2026 22:55:28 +0800 Subject: [PATCH] live: all-tasks foothold path (ALL_TASKS_FOOTHOLD, default off) Wire onlyMine=false pass/fail/running slices into candidate merge; exclude waiting; demote queue-only pending text; keep onlyMine queue gate. --- README.md | 2 + src/all_tasks_foothold.py | 213 +++++++++++++++++++++++++++++++ src/live_cycle.py | 26 +++- src/pending_pool.py | 4 +- tests/test_all_tasks_foothold.py | 125 ++++++++++++++++++ tests/test_live_cycle.py | 60 +++++++++ tests/test_pending_pool.py | 2 +- 7 files changed, 424 insertions(+), 8 deletions(-) create mode 100644 src/all_tasks_foothold.py create mode 100644 tests/test_all_tasks_foothold.py diff --git a/README.md b/README.md index 334a048..c1e3aec 100644 --- a/README.md +++ b/README.md @@ -87,3 +87,5 @@ tests/ # stdlib unittest 成功指标统一为 `verify_pass`（`verifyResult==1`，验证通过）；任务 status 为 `success` 但 `verifyResult==-1` 属于验证失败，积分为 0。 每轮 live 在凭证解析后记录一次 `[live] stats total=N queue=N verify_pass=N verify_fail=N failed=N unknown=N points=X`；读取失败则记录 `[live] stats unavailable reason=<安全原因>`，提交仍由原队列闸门决定。`GET /status` 保留凭证存在布尔值，新增 `success_metric: "verify_pass"` 和 `stats`：最近成功读取的上述数字及 `updated_at`（epoch 秒），尚未读取时为 null。queue 按未终结状态计数；积分仅累计 verify_pass 的有效数字。 + +`ALL_TASKS_FOOTHOLD=1` 可启用全部任务落点（默认 `0`）：只读取通过、失败、running 三桶，硬顶 3/3/2 页、每页 ≤100、单请求 ≤5 秒、总读取 ≤20 秒。任一桶错误整轮回退原候选，stderr 输出固定原因码与分页/截断计数。目标卡只扣除切片中已观察的卡；切片不保证全站覆盖、最新排序、已下载或零重复排队，仍保留原队列门控、marks 与 precheck。上限可用 `ALL_TASKS_MAX_PAGES_PASS/FAIL/RUNNING`、`ALL_TASKS_PAGE_SIZE` 调低，不能调高硬顶。 diff --git a/src/all_tasks_foothold.py b/src/all_tasks_foothold.py new file mode 100644 index 0000000..9b85754 --- /dev/null +++ b/src/all_tasks_foothold.py @@ -0,0 +1,213 @@ +"""Bounded all-scope evidence slices, isolated from my-queue state.""" +from dataclasses import dataclass, field +import json +import os +import re +import sys +import time +import urllib.parse +import urllib.request +from .config_params import GPU_ENUMS +from .live_queue import model_key, valid_total +from .pending_pool import classify_foothold +from .task_add_client import resolve_token + +TARGET_GPUS = ('Biren_166m', 'Sunrise_pt-200-x1', 'Iluvatar_bi-150', 'Iluvatar_mrv-100') +RANK = {'verified_reuse': 0, 'failed_foothold': 1, 'in_flight': 2} +REASONS = {'ok', 'missing/invalid_token', 'transport_error', 'http_error', + 'platform_error', 'malformed_data', 'filter_mismatch', 'deadline', + 'conflicting_task', 'reader_error'} + + +@dataclass +class FootholdSnapshot: + candidates: list = field(default_factory=list) + reason: str = 'ok' + buckets: dict = field(default_factory=dict) + + +def _limit(env, default): + try: + value = int(os.environ.get(env, str(default))) + return min(value, default) if value > 0 else default + except ValueError: + return default + + +def classify_api_record(record, bucket): + """Positive whitelist. A mismatch invalidates the complete snapshot.""" + if not isinstance(record, dict): + return None + status = record.get('status') + status = status.strip().lower() if isinstance(status, str) else None + result = record.get('verifyResult') + if type(result) is str and result.strip() in ('1', '-1', '0'): + result = int(result.strip()) + if result is not None and type(result) is not int: + return None + expected = {'pass': ('success', 1), 'fail': ('success', -1)} + if bucket in expected: + if (status, result) != expected[bucket]: + return None + elif bucket != 'running' or status != 'running' or result not in (None, 0): + return None + model, gpu = record.get('modelId'), record.get('gpuType') + if not isinstance(model, str) or not re.fullmatch(r'[\w.-]+/[\w./-]+', model_key(model)): + return None + if not isinstance(gpu, str) or not re.fullmatch(r'[\w .-]+', gpu.strip()): + return None + gpu = GPU_ENUMS.get(gpu.strip(), gpu.strip()) + return (model.strip(), gpu, {'pass': 'verified_reuse', 'fail': 'failed_foothold', 'running': 'in_flight'}[bucket]) + + +def build_candidates(evidence, *, target_gpus=TARGET_GPUS): + groups = {} + targets = list(dict.fromkeys(GPU_ENUMS[g] for g in target_gpus if g in GPU_ENUMS)) + for model, gpu, kind in evidence: + key = model_key(model) + row = groups.setdefault(key, dict(model_address=model, section='A', + foothold_class=kind, evidence_source='all-tasks', no_download=False, seen_gpus=[])) + if gpu not in row['seen_gpus']: + row['seen_gpus'].append(gpu) + if RANK[kind] < RANK[row['foothold_class']]: + row['foothold_class'] = kind + for row in groups.values(): + row['try_next_gpus'] = [g for g in targets if g not in row['seen_gpus']] + return sorted(groups.values(), key=candidate_priority) + + +def _kind(row): + text = row.get('foothold_text', '') + if text and row.get('evidence_source') != 'all-tasks': + return classify_foothold(text) + kind = row.get('foothold_class', 'unknown') + if kind == 'in_flight' and row.get('evidence_source') != 'all-tasks': + return 'unknown' + return kind + + +def candidate_priority(row): + return (RANK.get(_kind(row), 3), row.get('section', 'B') != 'A') + + +def _choices(row): + values = row['try_next_gpus'] if 'try_next_gpus' in row else [row.get('targetGpu')] + return list(dict.fromkeys(GPU_ENUMS[g] for g in values if isinstance(g, str) and g in GPU_ENUMS)) + + +def merge_candidates(base, dynamic): + """Keep base constraints; intersect explicit targets and subtract seen cards.""" + rows = {} + constrained = set() + for incoming in [*base, *dynamic]: + key = model_key(incoming.get('model_address') or incoming.get('modelAddress')) + row = dict(incoming) + row['foothold_class'] = _kind(row) + explicit_targets = 'try_next_gpus' in row or 'targetGpu' in row + row['try_next_gpus'] = _choices(row) + if key not in rows: + rows[key] = row + if explicit_targets: + constrained.add(key) + continue + existing = rows[key] + seen = set(existing.get('seen_gpus', [])) | set(row.get('seen_gpus', [])) + existing['seen_gpus'] = list(seen) + if key not in constrained and explicit_targets: + existing['try_next_gpus'] = row['try_next_gpus'] + constrained.add(key) + incoming_choices = row['try_next_gpus'] if explicit_targets else existing['try_next_gpus'] + existing['try_next_gpus'] = [g for g in existing['try_next_gpus'] if g in incoming_choices and g not in seen] + if candidate_priority(row)[0] < candidate_priority(existing)[0]: + existing['foothold_class'] = row['foothold_class'] + existing['evidence_source'] = row.get('evidence_source') + if row.get('skip_reason') and not existing.get('skip_reason'): + existing['skip_reason'] = row['skip_reason'] + return sorted(rows.values(), key=candidate_priority) + + +def load_all_tasks_foothold(*, opener=None, token_resolver=None, clock=None): + opener = opener or urllib.request.urlopen + clock = clock or time.monotonic + buckets = {} + def fail(reason): + return FootholdSnapshot(reason=reason, buckets=buckets) + try: + token = (token_resolver or resolve_token)() + except Exception: + return fail('missing/invalid_token') + deadline = clock() + 20 + size = _limit('ALL_TASKS_PAGE_SIZE', 100) + evidence, tasks = [], {} + for bucket, cap, filters in ( + ('pass', 3, {'verifyResult': 1, 'status': 'success'}), + ('fail', 3, {'verifyResult': -1, 'status': 'success'}), + ('running', 2, {'status': 'running'}), + ): + info = buckets[bucket] = dict(pages=0, valid=0, unique=0, truncated=False) + fingerprints = set() + raw_count = 0 + for page in range(1, _limit('ALL_TASKS_MAX_PAGES_' + bucket.upper(), cap) + 1): + remaining = deadline - clock() + if remaining <= 0: + return fail('deadline') + query = urllib.parse.urlencode(dict(onlyMine='false', pageSize=size, current=page, **filters)) + request = urllib.request.Request(os.environ.get('MODELHUB_API_BASE', 'https://modelhub.org.cn').rstrip('/') + '/api/adapt/task/page?' + query, + headers={'Xc-Token': token, 'Accept': 'application/json'}) + try: + with opener(request, timeout=min(5, remaining)) as response: + if not 200 <= response.status < 300: + return fail('http_error') + body = json.loads(response.read()) + except (ValueError, UnicodeError): + return fail('malformed_data') + except Exception: + return fail('transport_error') + if clock() >= deadline: + return fail('deadline') + if not isinstance(body, dict) or type(body.get('code')) not in (int, str) or str(body['code']) != '0': + return fail('platform_error') + data = body.get('data') + if not isinstance(data, dict) or not valid_total(data.get('total')): + return fail('malformed_data') + records = data.get('records') + if not isinstance(records, list) or len(records) > size: + return fail('malformed_data') + info['pages'] += 1 + page_evidence = [] + for record in records: + item = classify_api_record(record, bucket) + if item is None or any(token.lower() in part.lower() for part in item): + return fail('filter_mismatch') + task = record.get('taskId') + if task is not None and (type(task) not in (str, int) or not str(task).strip()): + return fail('malformed_data') + identity = ('task', str(task)) if task is not None else ('evidence', model_key(item[0]), item[1], item[2]) + normalized = (model_key(item[0]), *item[1:]) + if identity in tasks and tasks[identity] != normalized: + return fail('conflicting_task') + info['valid'] += 1 + page_evidence.append(normalized) + if identity not in tasks: + tasks[identity] = normalized + evidence.append(item) + info['unique'] += 1 + fingerprint = tuple(page_evidence) + raw_count += len(records) + info['truncated'] = bool(records) and raw_count < int(data['total']) + if fingerprint in fingerprints: + info['truncated'] = True + break + fingerprints.add(fingerprint) + if not records or not info['truncated']: + break + return FootholdSnapshot(build_candidates(evidence), buckets=buckets) + + +def report_snapshot(snapshot): + reason = snapshot.reason if snapshot.reason in REASONS else 'reader_error' + print('[foothold] reason=' + reason, file=sys.stderr) + for bucket in ('pass', 'fail', 'running'): + if bucket in snapshot.buckets: + info = snapshot.buckets[bucket] + print('[foothold] bucket=' + bucket + ' ' + ' '.join(f'{k}={int(info[k])}' for k in ('pages', 'valid', 'unique', 'truncated')), file=sys.stderr) diff --git a/src/live_cycle.py b/src/live_cycle.py index 6ee4719..f759e83 100644 --- a/src/live_cycle.py +++ b/src/live_cycle.py @@ -9,6 +9,8 @@ from .config_params import GPU_ENUMS, build_config_params_with_meta from .gate import evaluate_gate, queue_threshold from .live_queue import read_queue, read_queue_stats, model_key from .pending_pool import build_ordered_candidates +from .all_tasks_foothold import (load_all_tasks_foothold, merge_candidates, candidate_priority, + FootholdSnapshot, report_snapshot) from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source, token_probe_status from .ledger import load_baked from .marks import load_marks, REJECT_CODES, PERMANENT_PAIR_CODES @@ -19,12 +21,12 @@ _marked = {} _cycle_lock = threading.Lock() -def run_cycle(candidates=None, client=None, queue_reader=None, now=None, stats_reader=None, precheck=None): +def run_cycle(candidates=None, client=None, queue_reader=None, now=None, stats_reader=None, precheck=None, foothold_reader=None): with _cycle_lock: - return _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck) + return _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck, foothold_reader) -def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck): +def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck, foothold_reader): decisions = [] def emit(action, reason, **fields): @@ -90,7 +92,21 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck): except (OSError, ValueError, IndexError): emit('hold', 'candidate_read_failed') return decisions - candidates = sorted(candidates, key=lambda c: (c.get('section', 'B') != 'A', c.get('foothold_class') != 'verified_reuse')) + if os.environ.get('ALL_TASKS_FOOTHOLD', '0') == '1' and (foothold_reader is not None or queue_reader is None): + try: + snapshot = (foothold_reader or load_all_tasks_foothold)() + if not isinstance(snapshot, FootholdSnapshot): + raise ValueError + except Exception: + snapshot = FootholdSnapshot(reason='reader_error') + report_snapshot(snapshot) + dynamic = snapshot.candidates if snapshot.reason == 'ok' else [] + if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'): + section = os.environ.get('PENDING_SECTION', '').strip().upper() + if section: + dynamic = [row for row in dynamic if row.get('section') == section] + candidates = merge_candidates(candidates, dynamic) + candidates = sorted(candidates, key=candidate_priority) if not candidates: emit('hold', 'no_candidates') return decisions @@ -163,7 +179,7 @@ def _run_cycle(candidates, client, queue_reader, now, stats_reader, precheck): emit('hold', 'cycle_cap') break mapping = {} - choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')] + choices = candidate['try_next_gpus'] if 'try_next_gpus' in candidate else [candidate.get('targetGpu')] gpu = choose(choices, key, model, mapping) if gpu is None: emit('skip', 'all_gpus_done_or_unmapped', model=model) diff --git a/src/pending_pool.py b/src/pending_pool.py index 4c07d2f..981d893 100644 --- a/src/pending_pool.py +++ b/src/pending_pool.py @@ -46,12 +46,12 @@ def classify_foothold(text): """Classify literal pending evidence; mixed markers prefer verified.""" for label, markers in ( ('verified_reuse', ('已验证成功', '验证通过')), - ('in_flight', ('正在验证', '验证排队中', '验证进行中')), + ('in_flight', ('正在验证', '验证进行中')), ('failed_foothold', ('验证失败',)), ): if any(marker in text for marker in markers): return label - return 'unknown' + return 'queued_only' if '验证排队中' in text else 'unknown' def has_no_download_marker(text): diff --git a/tests/test_all_tasks_foothold.py b/tests/test_all_tasks_foothold.py new file mode 100644 index 0000000..9c71b5b --- /dev/null +++ b/tests/test_all_tasks_foothold.py @@ -0,0 +1,125 @@ +import io +import json +import os +import unittest +from unittest.mock import Mock, patch +from urllib.parse import parse_qs, urlparse +from src.all_tasks_foothold import (load_all_tasks_foothold, classify_api_record, + build_candidates, merge_candidates, candidate_priority, TARGET_GPUS, FootholdSnapshot) +from src import live_queue + + +def record(status='success', result=1, model='org/model', gpu='壁砺', **extra): + return dict(modelId=model, gpuType=gpu, status=status, verifyResult=result, **extra) + + +def response(records=(), total=None, **body): + result = Mock(status=200) + result.read.return_value = json.dumps(body or dict(code=0, data=dict(records=list(records), total=len(records) if total is None else total))).encode() + result.__enter__ = Mock(return_value=result) + result.__exit__ = Mock(return_value=False) + return result + + +class TestFoothold(unittest.TestCase): + def setUp(self): + self.env = patch.dict(os.environ, {}, clear=True) + self.env.start() + self.addCleanup(self.env.stop) + self.network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden')) + self.network.start() + self.addCleanup(self.network.stop) + + def load(self, responses, **kw): + opener = Mock(side_effect=responses) + snapshot = load_all_tasks_foothold(opener=opener, token_resolver=lambda: 'offline-credential', **kw) + return snapshot, opener + + def test_whitelist_rejects_waiting_conflicts_and_bad_types(self): + for status, result, bucket in [('waiting', 1, 'pass'), ('failed', -1, 'fail'), + ('running', 1, 'running'), ('success', True, 'pass'), + ('success', {}, 'pass'), ('other', 1, 'pass'), ('success', 1.0, 'pass')]: + self.assertIsNone(classify_api_record(record(status, result), bucket)) + for result in (1, '1'): + self.assertIsNotNone(classify_api_record(record(result=result), 'pass')) + for result in (None, 0, '0'): + self.assertIsNotNone(classify_api_record(record(' RUNNING ', result), 'running')) + for fields in (dict(modelId=True), dict(gpuType=''), dict(modelId='')): + self.assertIsNone(classify_api_record(dict(record(), **fields), 'pass')) + + def test_scope_buckets_aliases_order_and_queue_isolation(self): + before = live_queue.last_stats() + snap, opener = self.load([response([record()]), response([record(result=-1, model='org/fail')]), + response([record('running', None, model='org/run')])]) + self.assertEqual(snap.reason, 'ok') + self.assertEqual([r['foothold_class'] for r in snap.candidates], ['verified_reuse', 'failed_foothold', 'in_flight']) + self.assertEqual(snap.candidates[0]['try_next_gpus'], list(TARGET_GPUS[1:])) + for call in opener.call_args_list: + query = parse_qs(urlparse(call.args[0].full_url).query) + self.assertEqual(query['onlyMine'], ['false']) + self.assertNotIn('waiting', query.get('status', [])) + self.assertLessEqual(call.kwargs['timeout'], 5) + self.assertEqual(before, live_queue.last_stats()) + self.assertEqual(len(opener.call_args_list), 3) + + def test_any_bucket_error_discards_whole_snapshot(self): + for bad in (response([record('waiting', 1)]), response(data=None, code=0), + response(code=403), OSError('private-response')): + snap, _ = self.load([response([record()]), bad]) + self.assertNotEqual(snap.reason, 'ok') + self.assertEqual(snap.candidates, []) + snap, _ = self.load([response(), response(), response()]) + self.assertEqual(snap.reason, 'ok') + self.assertEqual(snap.candidates, []) + + def test_hard_caps_total_drift_and_invalid_env(self): + with patch.dict(os.environ, {'ALL_TASKS_MAX_PAGES_PASS': '999', 'ALL_TASKS_MAX_PAGES_FAIL': 'bad', + 'ALL_TASKS_MAX_PAGES_RUNNING': '-1', 'ALL_TASKS_PAGE_SIZE': '999'}): + pages = [response([record(model='org/p' + str(n))], total=1000+n) for n in range(3)] + pages += [response([record(result=-1, model='org/f' + str(n))], total=1000+n) for n in range(3)] + pages += [response([record('running', None, model='org/r' + str(n))], total=1000+n) for n in range(2)] + snap, opener = self.load(pages) + self.assertEqual(opener.call_count, 8) + self.assertTrue(all(b['truncated'] for b in snap.buckets.values())) + self.assertTrue(all('pageSize=100' in c.args[0].full_url for c in opener.call_args_list)) + snap, opener = self.load([], clock=Mock(side_effect=[0, 21])) + self.assertEqual(snap.reason, 'deadline') + opener.assert_not_called() + + def test_repeated_tasks_pages_and_conflicts(self): + r = record(taskId=1) + snap, opener = self.load([response([r], total=999), response([r], total=1000), response(), response()]) + self.assertEqual(snap.buckets['pass']['unique'], 1) + self.assertTrue(snap.buckets['pass']['truncated']) + self.assertEqual(opener.call_count, 4) + snap, _ = self.load([response([r]), response([record(result=-1, taskId=1)])]) + self.assertEqual(snap.reason, 'conflicting_task') + self.assertEqual(snap.candidates, []) + + def test_missing_token_never_opens(self): + opener = Mock() + snap = load_all_tasks_foothold(opener=opener, token_resolver=Mock(side_effect=ValueError())) + self.assertEqual(snap.reason, 'missing/invalid_token') + opener.assert_not_called() + + def test_merge_constraints_empty_targets_and_uniform_priority(self): + base = [dict(model_address='https://huggingface.co/ORG/Model/', framework='custom', + skip_reason='manual', section='B', foothold_class='verified_reuse', + try_next_gpus=['壁砺'], targetGpu='天垓150'), + dict(model_address='org/static', section='A', foothold_class='verified_reuse', try_next_gpus=['天垓150'])] + dynamic = build_candidates([('org/model', 'Biren_166m', 'in_flight'), ('org/run', 'MetaX_c-500', 'in_flight')]) + merged = merge_candidates(base, dynamic) + row = next(r for r in merged if 'huggingface' in r['model_address']) + self.assertEqual(row['framework'], 'custom') + self.assertEqual(row['skip_reason'], 'manual') + self.assertEqual(row['try_next_gpus'], []) + self.assertEqual(row['foothold_class'], 'verified_reuse') + self.assertEqual(len(merged), 3) + self.assertEqual(merged[-1]['model_address'], 'org/run') + self.assertEqual(merged[-1]['try_next_gpus'], list(TARGET_GPUS)) + self.assertEqual(candidate_priority(dict(foothold_class='in_flight'))[0], 3) + + def test_pending_queue_only_text(self): + from src.pending_pool import classify_foothold + self.assertEqual(classify_foothold('验证排队中'), 'queued_only') + self.assertEqual(classify_foothold('验证排队中; 验证通过'), 'verified_reuse') diff --git a/tests/test_live_cycle.py b/tests/test_live_cycle.py index 99964df..201797d 100644 --- a/tests/test_live_cycle.py +++ b/tests/test_live_cycle.py @@ -215,3 +215,63 @@ class TestLiveCycle(unittest.TestCase): def test_injected_queue_skips_stats_network(self): self.cycle() self.assertEqual(self.logs.getvalue().count('[live] stats unavailable reason=not_read'), 1) + + def test_foothold_explicit_disabled_and_injected_queue_stay_offline(self): + from src.all_tasks_foothold import FootholdSnapshot + dynamic = Mock(return_value=FootholdSnapshot()) + with patch.dict(os.environ, {'ALL_TASKS_FOOTHOLD': '1'}): + run_cycle([candidate('explicit')], self.client, lambda: (0, 'ok', set()), foothold_reader=dynamic) + dynamic.assert_not_called() + with patch('src.live_cycle.build_ordered_candidates', return_value=[candidate('base')]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline'}): + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=dynamic) + with patch.dict(os.environ, {'ALL_TASKS_FOOTHOLD': '1'}): + run_cycle(None, self.client, lambda: (0, 'ok', set())) + dynamic.assert_not_called() + + def test_foothold_merge_order_constraints_pending_and_fallback(self): + from src.all_tasks_foothold import FootholdSnapshot, build_candidates + base = [candidate('base', 'A', 'verified_reuse')] + snapshot = FootholdSnapshot(build_candidates([('org/dynamic', 'Biren_166m', 'in_flight')])) + with patch('src.live_cycle.build_ordered_candidates', return_value=base) as pending, patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'PENDING_SECTION': 'A', 'ALL_TASKS_FOOTHOLD': '1'}): + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot) + pending.assert_called_once_with() + self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/base', 'org/dynamic']) + _submitted.clear() + self.client.reset_mock() + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=Mock(side_effect=OSError('private-response'))) + self.assertEqual(self.client.submit.call_args.args[0]['modelAddress'], 'org/base') + self.assertIn('[foothold] reason=reader_error', self.logs.getvalue()) + self.assertNotIn('private-response', self.logs.getvalue()) + + def test_foothold_empty_intersection_never_revives_target(self): + from src.all_tasks_foothold import FootholdSnapshot, build_candidates + base = [dict(candidate('empty'), targetGpu='曦望')] + snapshot = FootholdSnapshot(build_candidates([('org/empty', 'Biren_166m', 'verified_reuse')])) + with patch('src.live_cycle.build_ordered_candidates', return_value=base), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'ALL_TASKS_FOOTHOLD': '1'}): + decisions = run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot) + self.client.submit.assert_not_called() + self.assertEqual(decisions[-1]['reason'], 'all_gpus_done_or_unmapped') + + def test_dynamic_still_respects_queue_marks_ledger_precheck(self): + from src.all_tasks_foothold import FootholdSnapshot + snapshot = FootholdSnapshot([dict(candidate('guarded'), evidence_source='all-tasks', foothold_class='verified_reuse')]) + with patch('src.live_cycle.build_ordered_candidates', return_value=[]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'ALL_TASKS_FOOTHOLD': '1'}): + for count in (95, None): + decisions = run_cycle(None, self.client, lambda: (count, 'ok', set()), foothold_reader=lambda: snapshot) + self.assertEqual(decisions[-1]['action'], 'hold') + with patch('src.live_cycle.load_baked', return_value=({('org/guarded', 'Biren_166m')}, set())): + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot) + _marked[('org/guarded', 'Biren_166m')] = 60028 + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot) + _marked.clear() + with patch.dict(os.environ, {'LIVE_PRECHECK': '1'}): + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: snapshot, + precheck=lambda _: ('ok', {'Biren_166m': 60027})) + self.client.submit.assert_not_called() + + def test_pending_section_b_does_not_gain_dynamic_section_a(self): + from src.all_tasks_foothold import FootholdSnapshot + dynamic = FootholdSnapshot([candidate('dynamic', 'A', 'verified_reuse')]) + with patch('src.live_cycle.build_ordered_candidates', return_value=[candidate('base')]), patch.dict(os.environ, {'PENDING_SUBMIT_MD': 'offline', 'PENDING_SECTION': 'B', 'ALL_TASKS_FOOTHOLD': '1'}): + run_cycle(None, self.client, lambda: (0, 'ok', set()), foothold_reader=lambda: dynamic) + self.assertEqual([c.args[0]['modelAddress'] for c in self.client.submit.call_args_list], ['org/base']) diff --git a/tests/test_pending_pool.py b/tests/test_pending_pool.py index 2088cb3..1a4fad5 100644 --- a/tests/test_pending_pool.py +++ b/tests/test_pending_pool.py @@ -59,7 +59,7 @@ class TestPendingPool(unittest.TestCase): 'org/fail', 'org/queue', 'org/progress', 'org/b']) self.assertEqual([r['foothold_class'] for r in rows], ['verified_reuse', 'verified_reuse', 'unknown', 'in_flight', - 'failed_foothold', 'in_flight', 'in_flight', 'unknown']) + 'failed_foothold', 'queued_only', 'in_flight', 'unknown']) for row in rows: self.assertEqual(row['evidence_source'], 'GAP' if row['foothold_class'] == 'unknown' else 'pending-foothold') self.assertEqual(rows[0]['foothold_text'], '他人:已验证成功; 正在验证')