fix: respect task page limits and recover stale capacity counts

README.md

@@ -2,6 +2,19 @@
 
 This repository is packaged for the ModelHub XC agent platform.
 
+## 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,14 +325,46 @@ class ModelHubClient:
         return active_count
 
     def count_active_tasks_exact(self) -> int:
-        """Use the platform's filtered totals instead of paging all history."""
+        """Conservative active occupancy from filtered, validated pages.
+
+        Totals can disagree with records. Count unique IDs, but retain the
+        larger reported total rather than inventing extra free slots.
+        """
         total = 0
+        seen: 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"]))
+            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 _request(
@@ -521,6 +553,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 +736,13 @@ 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)
+            print(f"[capacity] refresh_failed error={type(exc).__name__} code={code} count=unknown", flush=True)
             return None
 
     def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]:
@@ -779,6 +815,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"conservative_active={new_remote_count} details={details}", flush=True)
                     if new_remote_count > self._account_caps[index]:
                         self._account_caps[index] = new_remote_count
                         caps_changed = True
@@ -821,6 +862,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 +888,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 +988,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 +1051,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.27.1"

tests/test_capacity_refresh.py

@@ -0,0 +1,61 @@
+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, **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)]}}
+        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([97], pool.refresh_active_counts())
+        self.assertEqual(3, 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(3, pool.available_submit_slots())
+
+    def test_incomplete_or_repeated_pages_do_not_open_slots(self):
+        client = self.client(total=200, observed=100)
+        with self.assertRaises(ModelHubAPIError):
+            client.count_active_tasks_exact()
+        client.list_tasks_page = lambda **kwargs: {"data": {"records": []}}
+        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": [
+            {"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())

tests/test_concurrency.py

@@ -290,13 +290,16 @@ 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, **_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) * 100, min(current * 100, 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(["waiting", "waiting", "running"], calls)
 
     @staticmethod
     def monday_calibration_time(hour: int = 10) -> datetime: