live: only code==0 counts as submitted; log rejected w/ reason; 60027/60028 done, others skip per-run; 60007 halts; attempt cap 15 (70 tests)

src/live_cycle.py

@@ -12,7 +12,36 @@ from .pending_pool import build_ordered_candidates
 from .task_add_client import TaskAddClient, build_payload, resolve_token_with_source, token_probe_status
 from .ledger import load_baked, load_persisted, persist, rebuild
 
+REJECT_CODES = {
+    60006: 'CANNOT_CREATE_CONTEST_TASK_BECAUSE_NOT_IN_WHITE_LIST',
+    60007: 'MODEL_VERIFY_TASK_NUM_LIMIT',
+    60008: 'CREATE_CONTEST_TASK_REQUEST_EMPTY',
+    60009: 'CREATE_CONTEST_TASK_CONTRIBUTORS_EMPTY',
+    60010: 'CREATE_CONTEST_TASK_SUBMISSION_CONFIG_EMPTY',
+    60011: 'CREATE_CONTEST_TASK_GPU_TYPE_EMPTY',
+    60012: 'CREATE_CONTEST_TASK_MULTI_GPU_UNSUPPORTED',
+    60013: 'CREATE_CONTEST_TASK_MODEL_ID_EMPTY',
+    60014: 'CREATE_CONTEST_TASK_SCRIPT_SUBMIT_DISABLED',
+    60015: 'CREATE_CONTEST_TASK_TASK_TYPE_EMPTY',
+    60016: 'CREATE_CONTEST_TASK_FRAMEWORK_EMPTY',
+    60017: 'CREATE_CONTEST_TASK_DRIVER_IMAGE_NOT_FOUND',
+    60018: 'CREATE_CONTEST_TASK_CONTRIBUTOR_NOT_FOUND',
+    60019: 'CREATE_CONTEST_TASK_SUBMISSION_GPU_TYPE_MISMATCH',
+    60020: 'CREATE_CONTEST_TASK_MODEL_NOT_DOWNLOADED',
+    60021: 'CREATE_CONTEST_TASK_FRAMEWORK_UNRESOLVED',
+    60022: 'CREATE_CONTEST_TASK_GPU_CONFIG_UNSUPPORTED',
+    60023: 'CREATE_CONTEST_TASK_YAML_CONFIG_EMPTY',
+    60024: 'CREATE_CONTEST_TASK_YAML_CONFIG_INVALID',
+    60025: 'CREATE_CONTEST_TASK_GPU_TYPE_UNSUPPORTED',
+    60026: 'CREATE_CONTEST_TASK_MODEL_NOT_FOUND',
+    60027: 'CREATE_CONTEST_TASK_MODEL_ALREADY_VERIFIED',
+    60028: 'CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING',
+    60029: 'CREATE_CONTEST_TASK_TRANSFORMERS_REQUIRES_FAILED_BASELINE',
+    60030: 'CREATE_CONTEST_TASK_API_VERSION_UNSUPPORTED',
+}
+_DONE_CODES = {60027, 60028}
 _submitted = set()
+_rejected = set()
 _cycle_lock = threading.Lock()
 
 
@@ -50,6 +79,12 @@ def _run_cycle(candidates, client, queue_reader, now):
     except ValueError:
         emit('hold', 'invalid_cycle_cap')
         return decisions
+    try:
+        attempt_cap = int(os.environ.get('LIVE_MAX_ATTEMPTS_PER_CYCLE', '15'))
+        if attempt_cap < 1:
+            raise ValueError
+    except ValueError:
+        attempt_cap = 15
     if candidates is None:
         try:
             if os.environ.get('PENDING_SUBMIT_MD') or os.environ.get('PENDING_SUBMIT_TXT'):
@@ -67,13 +102,14 @@ def _run_cycle(candidates, client, queue_reader, now):
     try:
         persisted = load_persisted(state) | _submitted
         baked, blocked = load_baked()
-        known = persisted | baked
+        known = persisted | baked | _rejected
     except (OSError, ValueError, KeyError, TypeError):
         emit('hold', 'state_read_failed')
         return decisions
     reader = queue_reader or read_queue
     client = client if client is not None else TaskAddClient()
     attempts = 0
+    counted = 0
     for candidate in candidates:
         model = candidate.get('model_address') or candidate.get('modelAddress') or ''
         rule = banned_rule(model + ' ' + str(candidate.get('model_name', candidate.get('modelName', ''))))
@@ -90,7 +126,7 @@ def _run_cycle(candidates, client, queue_reader, now):
         if candidate.get('skip_reason'):
             emit('skip', 'pending_filter', model=model)
             continue
-        if attempts >= cap:
+        if counted >= cap:
             emit('hold', 'cycle_cap')
             break
         choices = candidate.get('try_next_gpus') or [candidate.get('targetGpu')]
@@ -142,6 +178,29 @@ def _run_cycle(candidates, client, queue_reader, now):
             reason = 'task_add_response' if status is not None else 'transport_error'
         except Exception:
             status, code, reason = None, None, 'submit_error'
+        if code is not None and code != 0:
+            emit('rejected', REJECT_CODES.get(code, 'unknown'), model=model,
+                 gpu=GPU_ENUMS[gpu], code=code, http=status, queue_before=count)
+            if code not in _DONE_CODES:
+                _rejected.add(pair)
+                persisted.discard(pair)
+                _submitted.discard(pair)
+                try:
+                    persist(state, persisted)
+                except OSError:
+                    emit('hold', 'state_write_failed')
+                    break
+            if code == 60007:
+                emit('hold', 'task_num_limit')
+                break
+            if attempts >= attempt_cap:
+                emit('hold', 'attempt_cap')
+                break
+            continue
+        counted += 1
         emit('submitted', reason, model=model, gpu=GPU_ENUMS[gpu], queue_before=count,
              http=status, code=code, **meta)
+        if attempts >= attempt_cap:
+            emit('hold', 'attempt_cap')
+            break
     return decisions

tests/test_live_cycle.py

@@ -1,9 +1,11 @@
 import io
+import json
+from pathlib import Path
 import os
 import tempfile
 import unittest
 from unittest.mock import Mock, patch
-from src.live_cycle import run_cycle, _submitted
+from src.live_cycle import run_cycle, _submitted, _rejected
 
 
 def candidate(name, section='B', foothold='unknown'):
@@ -18,6 +20,7 @@ class TestLiveCycle(unittest.TestCase):
         self.env.start()
         self.addCleanup(self.env.stop)
         _submitted.clear()
+        _rejected.clear()
         self.client = Mock()
         self.client.submit.return_value = dict(http_status=200, platform_code=0)
         network = patch('urllib.request.urlopen', side_effect=AssertionError('network forbidden'))
@@ -31,6 +34,80 @@ class TestLiveCycle(unittest.TestCase):
         reader = Mock(side_effect=[(n, 'platform_error' if n is None else 'ok', set()) for n in counts])
         return run_cycle(rows or [candidate('a')], self.client, reader)
 
+    def state_pairs(self):
+        return json.loads(Path(os.environ['LIVE_STATE_FILE']).read_text())
+
+    def test_done_rejection_keeps_reservation_and_continues(self):
+        for code in (60027, 60028):
+            with self.subTest(code=code):
+                self.client.submit.side_effect = [
+                    dict(http_status=200, platform_code=code),
+                    dict(http_status=200, platform_code=0)]
+                with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
+                    decisions = self.cycle([candidate(str(code)), candidate('ok' + str(code))], (1, 1))
+                self.assertEqual([d['action'] for d in decisions], ['rejected', 'submitted'])
+                self.assertIn('org/' + str(code) + '|Biren_166m', self.state_pairs())
+        self.assertIn('rejected', self.logs.getvalue())
+        self.assertIn('CREATE_CONTEST_TASK_MODEL_VERIFY_RUNNING', self.logs.getvalue())
+
+    def test_rejection_removed_but_not_retried_in_process(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60020)
+        self.cycle()
+        self.assertEqual(self.state_pairs(), [])
+        self.cycle(counts=())
+        self.client.submit.assert_called_once()
+
+    def test_unknown_rejection(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=99999)
+        decisions = self.cycle()
+        self.assertEqual(decisions[0]['action'], 'rejected')
+        self.assertEqual(decisions[0]['reason'], 'unknown')
+        self.assertEqual(self.state_pairs(), [])
+
+    def test_attempt_cap_and_invalid_fallback(self):
+        for value, expected in ((None, 15), ('bad', 15), ('0', 15), ('-1', 15), ('2', 2)):
+            with self.subTest(value=value):
+                _rejected.clear()
+                self.client.reset_mock()
+                self.client.submit.return_value = dict(http_status=200, platform_code=60020)
+                with patch.dict(os.environ, {} if value is None else {'LIVE_MAX_ATTEMPTS_PER_CYCLE': value}):
+                    decisions = self.cycle([candidate(str(i)) for i in range(expected)], (1,) * expected)
+                self.assertEqual(self.client.submit.call_count, expected)
+                self.assertEqual(decisions[-1]['reason'], 'attempt_cap')
+
+    def test_submit_cap_counts_successes_after_rejections(self):
+        self.client.submit.side_effect = [dict(http_status=200, platform_code=c) for c in (60020, 0, 60028, 0, 0)]
+        decisions = self.cycle([candidate(str(i)) for i in range(7)], (1,) * 5)
+        self.assertEqual(self.client.submit.call_count, 5)
+        self.assertEqual(sum(d['action'] == 'submitted' for d in decisions), 3)
+        self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
+
+    def test_task_limit_stops_cycle(self):
+        self.client.submit.return_value = dict(http_status=200, platform_code=60007)
+        decisions = self.cycle([candidate('a'), candidate('b')])
+        self.assertEqual([d['action'] for d in decisions], ['rejected', 'hold'])
+        self.assertEqual(decisions[-1]['reason'], 'task_num_limit')
+        self.assertEqual(self.state_pairs(), [])
+        self.client.submit.assert_called_once()
+
+    def test_unknown_outcomes_count_and_reserve_before_submit(self):
+        for result in (dict(http_status=200, platform_code='0'), dict(http_status=None, platform_code=None), OSError()):
+            with self.subTest(result=result):
+                _submitted.clear()
+                Path(os.environ['LIVE_STATE_FILE']).unlink(missing_ok=True)
+                self.client.reset_mock()
+                def submit(payload):
+                    self.assertIn('org/a|Biren_166m', self.state_pairs())
+                    if isinstance(result, Exception):
+                        raise result
+                    return result
+                self.client.submit.side_effect = submit
+                with patch.dict(os.environ, {'LIVE_MAX_SUBMITS_PER_CYCLE': '1'}):
+                    decisions = self.cycle([candidate('a'), candidate('b')])
+                self.assertEqual(decisions[0]['action'], 'submitted')
+                self.assertEqual(decisions[-1]['reason'], 'cycle_cap')
+                self.client.submit.assert_called_once()
+
     def test_high_and_failed_queue(self):
         for n in (95, None):
             decisions = self.cycle(counts=(n,))