fix: verify capacity from dual paginated active-task scans

README.md

@@ -2,6 +2,35 @@
 
 This repository is packaged for the ModelHub XC agent platform.
 
+## 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
+filtered waiting/running pagination, validate responses and deduplicate task IDs.
+When reported totals disagree with returned rows, occupancy uses the larger
+value conservatively and logs the discrepancy. Missing or non-progressing pages
+fail closed. Failed reads report `active_counts=unknown` and
+`reason=capacity_read_unavailable`, retaining the last-success age for diagnosis;
+they do not advertise cached occupancy as a freshly confirmed full queue.
+Healthy accounts remain usable, and the next successful refresh reopens slots.
+The five-slot recent-model reserve remains in force.
+
 ## Single-worker takeover release 2026.10.26.1
 
 This release is intended to replace the competing workers, using all configured

modelhub_submmit_api/market_intelligence.py

@@ -410,7 +410,7 @@ class MarketIntelligenceManager:
         }
         payload = client.list_tasks_page(
             current=1,
-            page_size=200,
+            page_size=100,
             only_mine=False,
             begin_time=cutoff,
             end_time=now,

modelhub_submmit_api/modelhub_client.py

@@ -151,7 +151,7 @@ class ModelHubClient:
             "/api/adapt/task/page",
             query={
                 "current": current,
-                "pageSize": page_size,
+                "pageSize": max(1, min(100, int(page_size))),
                 "onlyMine": str(only_mine).lower(),
                 "beginTime": query_time(begin_time),
                 "endTime": query_time(end_time),
@@ -325,15 +325,63 @@ class ModelHubClient:
         return active_count
 
     def count_active_tasks_exact(self) -> int:
-        """Use the platform's filtered totals instead of paging all history."""
-        total = 0
+        """Confirm active IDs with two independently paginated filtered reads.
+
+        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.
+        """
+        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"):
-            page = self.list_tasks_page(current=1, page_size=1, only_mine=True, status=status)
-            data = page.get("data")
-            if not isinstance(data, dict) or "total" not in data:
-                raise ModelHubAPIError(f"Active-task total is unavailable for status={status}")
-            total += max(0, int(data["total"]))
-        return total
+                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,
@@ -521,6 +569,7 @@ class ModelHubClientPool:
         # persisted capacity is not enough without a successful active-count
         # read for the current scheduling window.
         self._count_known: list[bool] = [False for _ in clients]
+        self._count_success_at = [0.0 for _ in clients]
         self._active_refresh_at: float = 0.0
         self._counts_initialized = False
         self._state_lock = threading.Lock()
@@ -703,10 +752,14 @@ class ModelHubClientPool:
 
     def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int | None) -> int | None:
         try:
-            if max_count is None and hasattr(client, "count_active_tasks_exact"):
+            if hasattr(client, "count_active_tasks_exact"):
                 return client.count_active_tasks_exact()
-            return client.count_active_tasks(max_count=max_count, page_size=200)
-        except Exception:
+            return client.count_active_tasks(max_count=max_count, page_size=100)
+        except Exception as exc:
+            # Never log request URLs/headers or credentials.
+            code = getattr(exc, "code", None)
+            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]]:
@@ -779,6 +832,11 @@ class ModelHubClientPool:
                     old_remote_count = self._remote_counts[index]
                     new_remote_count = max(0, int(count))
                     self._count_known[index] = True
+                    self._count_success_at[index] = refreshed_at
+                    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"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
@@ -821,6 +879,7 @@ class ModelHubClientPool:
             1
             for index in range(len(self.clients))
             if index not in self._capacity_probe_attempted
+            and self._count_known[index]
             and self._capacity_probe_cycle >= self._capacity_probe_cooldown_until[index]
             and self._effective_count_locked(index) >= self._account_caps[index]
         )
@@ -846,9 +905,20 @@ class ModelHubClientPool:
             normal_slots = sum(
                 max(0, self._account_caps[index] - self._effective_count_locked(index))
                 for index in range(len(self.clients))
+                if self._count_known[index]
             )
             return normal_slots if normal_slots > 0 else self._probe_slot_count_locked()
 
+    def active_count_status(self) -> list[dict[str, Any]]:
+        self._refresh_active_counts()
+        with self._state_lock:
+            now = time.monotonic()
+            return [{"account": i + 1, "known": self._count_known[i],
+                     "active": self._effective_count_locked(i) if self._count_known[i] else None,
+                     "lastSuccessAgeSeconds": round(now - self._count_success_at[i], 1)
+                     if self._count_success_at[i] else None}
+                    for i in range(len(self.clients))]
+
     def list_tasks(self, **kwargs):  # noqa: ANN003, ANN001
         # Default: use only the reader client to avoid fanout amplification.
         # The fanout_all parameter allows explicit cross-account merging when needed.
@@ -935,6 +1005,7 @@ class ModelHubClientPool:
                 index: self._account_caps[index] - self._effective_count_locked(index)
                 for index in range(len(self.clients))
                 if index not in excluded
+                and self._count_known[index]
                 and (weekly_target is None or index == weekly_target)
                 and (
                     reserve_capacity_slots is None
@@ -997,6 +1068,7 @@ class ModelHubClientPool:
                 index
                 for index in range(len(self.clients))
                 if index not in excluded
+                and self._count_known[index]
                 and (weekly_target is None or index == weekly_target)
                 and (weekly_target is not None or index not in self._capacity_probe_attempted)
                 and (

modelhub_submmit_api/poll_runner.py

@@ -1056,8 +1056,12 @@ def run_poll_loop(
             if hasattr(modelhub_client, "weekly_calibration_status") else {}
         )
         available_slots = modelhub_client.available_submit_slots() if hasattr(modelhub_client, "available_submit_slots") else None
+        count_status = modelhub_client.active_count_status() if hasattr(modelhub_client, "active_count_status") else []
+        display_counts = [str(item["active"]) if item["known"] else "unknown" for item in count_status]
+        if count_status and any(not item["known"] for item in count_status):
+            log(f"[capacity] account_status={count_status} unknown_accounts=deferred")
         log(
-            f"[poll] cycle={cycles} active_counts={','.join(str(count) for count in active_counts) if active_counts else 'n/a'} "
+            f"[poll] cycle={cycles} active_counts={','.join(display_counts or [str(count) for count in active_counts]) or 'n/a'} "
             f"capacity_limits={','.join(str(limit) for limit in capacity_limits) if capacity_limits else 'n/a'} "
             f"capacity_probe={'on' if capacity_probe else 'off'} "
             f"weekly_calibration={'on' if calibration.get('targetAccount') else 'off'} "
@@ -1084,7 +1088,8 @@ def run_poll_loop(
                     state_sync.healthy = False
                     state_sync.last_error = str(exc)
                     state_sync.write_readiness(ready=False, reason="state_sync_unhealthy")
-            log(f"[poll] cycle={cycles} sleep={base_args.poll_interval_seconds}s reason=no_available_slots")
+            no_slot_reason = "capacity_read_unavailable" if any(not item["known"] for item in count_status) else "no_available_slots"
+            log(f"[poll] cycle={cycles} sleep={base_args.poll_interval_seconds}s reason={no_slot_reason}")
             time.sleep(base_args.poll_interval_seconds)
             continue
 

modelhub_submmit_api/version.py

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

tests/test_capacity_refresh.py

@@ -0,0 +1,122 @@
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "modelhub_submmit_api"))
+from modelhub_client import ModelHubClient, ModelHubClientPool, ModelHubAPIError
+
+
+class CapacityRefreshTests(unittest.TestCase):
+    def test_api_boundary_clamps_all_callers(self):
+        client = ModelHubClient(token="test")
+        with patch.object(client, "_request", return_value={}) as request:
+            client.list_tasks_page(page_size=200)
+            self.assertEqual(100, request.call_args.kwargs["query"]["pageSize"])
+
+    def client(self, total=97, observed=94):
+        client = ModelHubClient(token="test")
+        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(
+                    (current - 1) * page_size, min(current * page_size, observed if status == "waiting" else 0))]}}
+        client.list_tasks_page = page
+        return client
+
+    def test_full_cache_refreshes_to_free_slots_with_disagreeing_total(self):
+        client = self.client()
+        pool = ModelHubClientPool([client], active_task_cap=100)
+        pool._remote_counts[0] = 100
+        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):
+        client = self.client()
+        pool = ModelHubClientPool([client], active_task_cap=100)
+        pool.refresh_active_counts()
+        with patch.object(client, "list_tasks_page", side_effect=ModelHubAPIError("分页大小不能超过100")):
+            pool.refresh_active_counts()
+            self.assertEqual(0, pool.available_submit_slots())
+            self.assertIsNone(pool.active_count_status()[0]["active"])
+            self.assertIsNone(pool._reserve_account(set()))
+        pool.refresh_active_counts()
+        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": {}}
+        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": [] 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,13 +290,17 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
         client = ModelHubClient(token="test-token")
         calls: list[str] = []
 
-        def page(*, status: str, **_kwargs) -> dict:  # noqa: ANN003
+        def page(*, status: str, current: int = 1, page_size: int = 100, **_kwargs) -> dict:  # noqa: ANN003
             calls.append(status)
-            return {"data": {"total": {"waiting": 175, "running": 25}[status]}}
+            total = {"waiting": 175, "running": 25}[status]
+            return {"data": {"total": total, "records": [
+                {"taskId": f"{status}-{i}", "status": status}
+                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", "running"], calls)
+        self.assertEqual(8, calls.count("waiting"))
+        self.assertEqual(4, calls.count("running"))
 
     @staticmethod
     def monday_calibration_time(hour: int = 10) -> datetime: