fix: verify capacity from dual paginated active-task scans

README.md

@@ -2,7 +2,23 @@
 
 This repository is packaged for the ModelHub XC agent platform.
 
-## Capacity refresh fix 2026.10.27.1
+## Capacity pagination verification 2026.10.28.1
+
+Supersedes the 2026.10.27.1 policy of using the larger reported total.
+The platform can return inflated totals, incorrect page counts, and short pages.
+Capacity now reads waiting/running pages through an explicit empty page, ignores
+total/pages for termination, and deduplicates task IDs. It repeats the scan with
+page sizes 100 and 50; only matching active ID sets confirm occupancy. Reported
+totals are diagnostic only. Repeated pages, malformed records, request failures,
+changing snapshots or the 100-page-per-status budget fail closed for that account.
+This costs additional read requests but scans active tasks, not full history.
+Two agreeing reads are not an atomic snapshot or a guarantee against a platform
+omission shared by both queries; server submission limits remain authoritative.
+Logs report `method=dual_page_scan`, `confirmed_active`, and safe failure reasons.
+No age cleanup, routing, success-rate gates, recent-five-slot reserve, or weekly
+capacity exploration policy changes. Existing tasks are not cancelled.
+
+## Previous capacity refresh fix 2026.10.27.1
 
 All task-page requests are clamped to the platform's 100-record limit, including
 community sampling. Normal and weekly capacity refreshes now use the same

modelhub_submmit_api/modelhub_client.py

@@ -325,47 +325,63 @@ class ModelHubClient:
         return active_count
 
     def count_active_tasks_exact(self) -> int:
-        """Conservative active occupancy from filtered, validated pages.
+        """Confirm active IDs with two independently paginated filtered reads.
 
-        Totals can disagree with records. Count unique IDs, but retain the
-        larger reported total rather than inventing extra free slots.
+        The platform's total/pages fields can describe a different population.
+        Only an explicit empty page terminates a scan, never those fields or a
+        short page. Two page sizes must agree before advertising any free slots.
+        This is not an atomic snapshot: concurrent queue changes fail closed.
         """
-        total = 0
-        seen: set[str] = set()
-        details = []
-        for status in ("waiting", "running"):
-            current, reported = 1, 0
-            status_ids: set[str] = set()
-            while True:
-                data = self.list_tasks_page(current=current, page_size=100, only_mine=True, status=status).get("data")
-                if not isinstance(data, dict) or "total" not in data or not isinstance(data.get("records"), list):
-                    raise ModelHubAPIError(f"Incomplete active-task response for status={status}")
-                count = int(data["total"])
-                if count < 0:
-                    raise ModelHubAPIError("Negative active-task total")
-                reported = max(reported, count)
-                pages = max(1, int(data.get("pages") or 0), (count + 99) // 100)
-                rows = data["records"]
-                prior = len(status_ids)
-                for row in rows:
-                    if not isinstance(row, dict) or not row.get("taskId") or row.get("status") != status:
-                        raise ModelHubAPIError(f"Invalid filtered active-task record for status={status}")
-                    status_ids.add(str(row["taskId"]))
-                if current >= pages:
-                    if current > 1 and len(status_ids) == prior:
-                        raise ModelHubAPIError("Active-task pagination made no progress")
-                    break
-                if not rows or len(status_ids) == prior or current >= 1000:
-                    raise ModelHubAPIError("Incomplete active-task pagination")
-                current += 1
-            # An ID can move from waiting to running between requests.
-            overlap = len(status_ids & seen)
-            total += max(reported, len(status_ids)) - overlap
-            seen.update(status_ids)
-            details.append({"status": status, "reported": reported, "unique": len(status_ids),
-                            "mismatch": reported != len(status_ids)})
-        self.active_count_details = details
-        return total
+        def invalid(reason: str) -> ModelHubAPIError:
+            error = ModelHubAPIError(reason)
+            error.capacity_reason = reason
+            return error
+
+        def scan(page_size: int) -> tuple[set[str], list[dict[str, Any]]]:
+            active: set[str] = set()
+            details = []
+            for status in ("waiting", "running"):
+                ids: set[str] = set()
+                reported = None
+                for current in range(1, 101):
+                    response = self.list_tasks_page(
+                        current=current, page_size=page_size, only_mine=True, status=status,
+                    )
+                    data = response.get("data") if isinstance(response, dict) else None
+                    if not isinstance(data, dict) or not isinstance(data.get("records"), list):
+                        raise invalid("invalid_active_response")
+                    try:
+                        value = int(data["total"])
+                        if value >= 0:
+                            reported = max(reported or 0, value)
+                    except (KeyError, ValueError, TypeError):
+                        pass  # Metadata is diagnostic, not an occupancy source.
+                    rows = data["records"]
+                    if not rows:
+                        break
+                    if len(rows) > page_size:
+                        raise invalid("page_size_not_respected")
+                    previous = len(ids)
+                    for row in rows:
+                        if not isinstance(row, dict) or not row.get("taskId") or row.get("status") != status:
+                            raise invalid("invalid_active_record")
+                        ids.add(str(row["taskId"]))
+                    if len(ids) == previous:
+                        raise invalid("active_pagination_no_progress")
+                else:
+                    raise invalid("active_pagination_budget_exhausted")
+                active.update(ids)
+                details.append({"status": status, "reported": reported, "unique": len(ids),
+                                "mismatch": reported != len(ids), "pageSize": page_size})
+            return active, details
+
+        self.active_count_details = []
+        first, details = scan(100)
+        confirmed, confirmation_details = scan(50)
+        if first != confirmed:
+            raise invalid("active_snapshot_changed")
+        self.active_count_details = details + confirmation_details
+        return len(confirmed)
 
     def _request(
         self,
@@ -742,7 +758,8 @@ class ModelHubClientPool:
         except Exception as exc:
             # Never log request URLs/headers or credentials.
             code = getattr(exc, "code", None)
-            print(f"[capacity] refresh_failed error={type(exc).__name__} code={code} count=unknown", flush=True)
+            reason = getattr(exc, "capacity_reason", "request_failed")
+            print(f"[capacity] refresh_failed error={type(exc).__name__} code={code} reason={reason} count=unknown", flush=True)
             return None
 
     def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]:
@@ -819,7 +836,7 @@ class ModelHubClientPool:
                     details = getattr(self.clients[index], "active_count_details", [])
                     if any(item.get("mismatch") for item in details):
                         print(f"[capacity] account={index + 1:02d} total_records_mismatch=true "
-                              f"conservative_active={new_remote_count} details={details}", flush=True)
+                              f"confirmed_active={new_remote_count} method=dual_page_scan details={details}", flush=True)
                     if new_remote_count > self._account_caps[index]:
                         self._account_caps[index] = new_remote_count
                         caps_changed = True

modelhub_submmit_api/version.py

@@ -1 +1 @@
-AGENT_VERSION = "2026.10.27.1"
+AGENT_VERSION = "2026.10.28.1"

tests/test_capacity_refresh.py

@@ -15,10 +15,11 @@ class CapacityRefreshTests(unittest.TestCase):
 
     def client(self, total=97, observed=94):
         client = ModelHubClient(token="test")
-        def page(*, status, page_size, **kwargs):
+        def page(*, status, page_size, current=1, **kwargs):
             self.assertLessEqual(page_size, 100)
             return {"data": {"total": total if status == "waiting" else 0, "records": [
-                {"taskId": str(i), "status": status} for i in range(observed if status == "waiting" else 0)]}}
+                {"taskId": str(i), "status": status} for i in range(
+                    (current - 1) * page_size, min(current * page_size, observed if status == "waiting" else 0))]}}
         client.list_tasks_page = page
         return client
 
@@ -26,8 +27,8 @@ class CapacityRefreshTests(unittest.TestCase):
         client = self.client()
         pool = ModelHubClientPool([client], active_task_cap=100)
         pool._remote_counts[0] = 100
-        self.assertEqual([97], pool.refresh_active_counts())
-        self.assertEqual(3, pool.available_submit_slots())
+        self.assertEqual([94], pool.refresh_active_counts())
+        self.assertEqual(6, pool.available_submit_slots())
         self.assertTrue(client.active_count_details[0]["mismatch"])
 
     def test_error_is_unknown_and_recovery_reopens_slots(self):
@@ -40,22 +41,82 @@ class CapacityRefreshTests(unittest.TestCase):
             self.assertIsNone(pool.active_count_status()[0]["active"])
             self.assertIsNone(pool._reserve_account(set()))
         pool.refresh_active_counts()
-        self.assertEqual(3, pool.available_submit_slots())
+        self.assertEqual(6, pool.available_submit_slots())
 
     def test_incomplete_or_repeated_pages_do_not_open_slots(self):
         client = self.client(total=200, observed=100)
+        client.list_tasks_page = lambda **kwargs: {"data": {"records": [
+            {"taskId": "same", "status": kwargs["status"]}]}}
         with self.assertRaises(ModelHubAPIError):
             client.count_active_tasks_exact()
-        client.list_tasks_page = lambda **kwargs: {"data": {"records": []}}
+        client.list_tasks_page = lambda **kwargs: {"data": {}}
         with self.assertRaises(ModelHubAPIError):
             client.count_active_tasks_exact()
 
     def test_duplicate_ids_and_transition_are_counted_once(self):
         client = ModelHubClient(token="test")
-        client.list_tasks_page = lambda **kwargs: {"data": {"total": 1, "records": [
+        client.list_tasks_page = lambda **kwargs: {"data": {"total": 1, "records": [] if kwargs['current'] > 1 else [
             {"taskId": "same", "status": kwargs["status"]},
             {"taskId": "same", "status": kwargs["status"]}]}}
         self.assertEqual(1, client.count_active_tasks_exact())
 
     def test_total_lower_than_observed_uses_larger_count(self):
         self.assertEqual(99, self.client(total=97, observed=99).count_active_tasks_exact())
+
+    def test_inflated_total_does_not_hide_capacity(self):
+        self.assertEqual(100, self.client(total=773, observed=100).count_active_tasks_exact())
+
+    def test_underreported_pages_do_not_truncate_scan(self):
+        client = self.client(total=58, observed=175)
+        original = client.list_tasks_page
+        def page(**kwargs):
+            result = original(**kwargs)
+            result['data']['pages'] = 1
+            return result
+        client.list_tasks_page = page
+        self.assertEqual(175, client.count_active_tasks_exact())
+
+    def test_different_confirmation_ids_fail_closed_even_with_equal_counts(self):
+        client = self.client(total=1, observed=1)
+        original = client.list_tasks_page
+        def page(**kwargs):
+            result = original(**kwargs)
+            if kwargs['page_size'] == 50:
+                for row in result['data']['records']:
+                    row['taskId'] = 'replacement'
+            return result
+        client.list_tasks_page = page
+        with self.assertRaisesRegex(ModelHubAPIError, 'active_snapshot_changed'):
+            client.count_active_tasks_exact()
+
+    def test_confirmation_error_does_not_open_slots(self):
+        client = self.client(total=773, observed=94)
+        original = client.list_tasks_page
+        def page(**kwargs):
+            if kwargs['page_size'] == 50:
+                raise ModelHubAPIError('temporary failure')
+            return original(**kwargs)
+        client.list_tasks_page = page
+        pool = ModelHubClientPool([client], active_task_cap=100)
+        pool.refresh_active_counts()
+        self.assertEqual(0, pool.available_submit_slots())
+        self.assertFalse(pool.active_count_status()[0]['known'])
+
+    def test_empty_queue_confirmed_despite_stale_total(self):
+        self.assertEqual(0, self.client(total=773, observed=0).count_active_tasks_exact())
+
+    def test_short_page_is_not_end_of_scan(self):
+        client = ModelHubClient(token='test')
+        def page(**kwargs):
+            rows = [{'taskId': str(kwargs['current']), 'status': 'waiting'}] if (
+                kwargs['status'] == 'waiting' and kwargs['current'] <= 3) else []
+            return {'data': {'records': rows}}
+        client.list_tasks_page = page
+        self.assertEqual(3, client.count_active_tasks_exact())
+
+    def test_scan_budget_is_bounded(self):
+        client = ModelHubClient(token='test')
+        client.list_tasks_page = lambda **kw: {'data': {'records': [
+            {'taskId': str(kw['current']), 'status': kw['status']}]}}
+        with self.assertRaisesRegex(ModelHubAPIError, 'active_pagination_budget_exhausted'):
+            client.count_active_tasks_exact()

tests/test_concurrency.py

@@ -290,16 +290,17 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
         client = ModelHubClient(token="test-token")
         calls: list[str] = []
 
-        def page(*, status: str, current: int = 1, **_kwargs) -> dict:  # noqa: ANN003
+        def page(*, status: str, current: int = 1, page_size: int = 100, **_kwargs) -> dict:  # noqa: ANN003
             calls.append(status)
             total = {"waiting": 175, "running": 25}[status]
             return {"data": {"total": total, "records": [
                 {"taskId": f"{status}-{i}", "status": status}
-                for i in range((current - 1) * 100, min(current * 100, total))]}}
+                for i in range((current - 1) * page_size, min(current * page_size, total))]}}
 
         client.list_tasks_page = page  # type: ignore[method-assign]
         self.assertEqual(200, client.count_active_tasks_exact())
-        self.assertEqual(["waiting", "waiting", "running"], calls)
+        self.assertEqual(8, calls.count("waiting"))
+        self.assertEqual(4, calls.count("running"))
 
     @staticmethod
     def monday_calibration_time(hour: int = 10) -> datetime: