feat: pack recent submissions into reserved account slots first

Preserve shared queue positions for old-model candidates with constrained
admission. Retain headroom balancing for ties and weekly calibration pinning.
Add offline mixed-age allocation regressions.

README.md

@@ -301,6 +301,26 @@ model/GPU combinations, the same run progressively expands discovery to the last
 7 days, the last 30 days, and finally older history (up to 3,000 models). Scanning
 stops as soon as enough replacement candidates have been found.
 
+When only recent-model reserved positions remain, discovery also deepens the
+same recent window up to the configured scan cap (3,000 by default). It never
+admits older models into those positions or lowers the quality gates. This avoids
+rechecking only an exhausted short prefix on every poll.
+
+An ordinary account-capacity rejection temporarily marks that account full and
+tries the next account. It does not lower the persisted capacity using a cached
+active count: another worker may have filled the queue since that count was
+read. Downward capacity learning remains confined to weekly calibration.
+
+Account allocation is age-aware: recent models first use accounts with only
+reserved positions left, then the account with the fewest shared positions
+remaining. Old models continue to use the account with the most eligible space.
+Largest-headroom and rotating tie-breaks balance otherwise equivalent accounts;
+weekly calibration remains pinned to its target account. For example, with
+accounts at 94/100 and 95/100 and five recent-only positions each, assigning a
+recent model to the second account preserves the first account's last old-model
+position. A recent-then-old pair can both be accepted rather than stranding the
+old model after the recent model takes its only eligible position.
+
 Model verification results are reused for 15 minutes across poll cycles, and a
 locally failed model/GPU pair cools down for 24 hours instead of being excluded
 forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done`

modelhub_submmit_api/main.py

@@ -822,6 +822,17 @@ def build_adaptive_scan_stages(
                 "limit": max_models,
             }
         )
+    if not allow_older_than_recent_window:
+        # Reserved positions still need a deep search of eligible recent
+        # models. Widening the date window alone repeatedly scans the same
+        # short prefix (often already processed) on every poll cycle.
+        stages.append(
+            {
+                "name": "recent_window_full",
+                "updatedAfter": recent_cutoff,
+                "limit": max_models,
+            }
+        )
     deduped: list[dict[str, Any]] = []
     seen: set[tuple[str | None, int, str]] = set()
     for stage in stages:

modelhub_submmit_api/modelhub_client.py

@@ -948,6 +948,26 @@ class ModelHubClientPool:
             usable = {index: remaining for index, remaining in remaining_by_index.items() if remaining > 0}
             if not usable:
                 return None
+            if (
+                reserve_capacity_slots is None
+                and weekly_target is None
+                and self.recent_model_reserve_slots > 0
+            ):
+                # Recent models can use every position; old models cannot.
+                # Pack recent work into reserved positions first, then the
+                # account nearest its old-model threshold. This preserves
+                # shared positions for later old candidates, regardless of
+                # candidate arrival order. Keep largest-headroom/round-robin
+                # balancing among accounts with the same shared capacity.
+                shared_slots = {
+                    index: max(0, remaining - self.recent_model_reserve_slots)
+                    for index, remaining in usable.items()
+                }
+                least_shared = min(shared_slots.values())
+                usable = {
+                    index: remaining for index, remaining in usable.items()
+                    if shared_slots[index] == least_shared
+                }
             best_remaining = max(usable.values())
             tied = [index for index, remaining in usable.items() if remaining == best_remaining]
             selected_index = min(
@@ -1029,7 +1049,12 @@ class ModelHubClientPool:
                 self._capacity_probe_cooldown_until[index] = (
                     self._capacity_probe_cycle + self._capacity_probe_cooldown_cycles
                 )
-            else:
+            elif self._weekly_target_locked() == index:
+                # Only the scheduled calibration may lower a learned cap.
+                # Ordinary rejection can race with another agent filling this
+                # account; our cached count is not evidence of its capacity.
+                # Persisting it (even zero -> one) strands otherwise usable
+                # positions until the next calibration.
                 observed_capacity = max(1, self._effective_count_locked(index))
                 if observed_capacity < self._account_caps[index]:
                     self._account_caps[index] = observed_capacity

tests/test_concurrency.py

@@ -6,6 +6,7 @@ import tempfile
 import threading
 import unittest
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from unittest.mock import patch
@@ -216,6 +217,75 @@ def make_candidate(index: int) -> dict:
 
 
 class ClientPoolConcurrencyTests(unittest.TestCase):
+    def test_recent_work_preserves_shared_slots_for_later_old_models(self) -> None:
+        for initial_counts, ages, expected_counts in (
+            ([94, 95], ["recent", "old"], [95, 96]),
+            ([93, 94], ["recent", "recent", "old", "old"], [95, 96]),
+        ):
+            with self.subTest(initial_counts=initial_counts):
+                clients = [FakeClient(active_count=count) for count in initial_counts]
+                pool = ModelHubClientPool(
+                    clients,  # type: ignore[arg-type]
+                    active_task_cap=100,
+                    recent_model_reserve_slots=5,
+                    active_counts_ttl=60,
+                    capacity_probe_interval_cycles=0,
+                )
+                for index, age in enumerate(ages):
+                    pool.add_task_for_model(
+                        {"model": f"model-{index}"},
+                        model_last_modified=(
+                            "2026-10-01T00:00:00Z" if age == "recent"
+                            else "2025-01-01T00:00:00Z"
+                        ),
+                        submitted_at="2026-10-02T00:00:00Z",
+                    )
+                self.assertEqual(expected_counts, pool.active_task_counts())
+                self.assertEqual(len(ages), sum(len(client.submitted) for client in clients))
+                self.assertEqual(0, pool.old_model_submit_slots())
+
+    def test_recent_packing_keeps_largest_headroom_without_reserved_slots(self) -> None:
+        clients = [FakeClient(active_count=2), FakeClient(active_count=3)]
+        pool = ModelHubClientPool(
+            clients,  # type: ignore[arg-type]
+            active_task_cap=5,
+            recent_model_reserve_slots=0,
+        )
+        pool.add_task({"model": "recent"})
+        self.assertEqual([1, 0], [len(client.submitted) for client in clients])
+
+    def test_stale_capacity_rejection_does_not_poison_learned_limit(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            state_path = Path(temporary_dir) / "capacity.json"
+            client = FakeClient(active_count=0, reject_capacity=True)
+            pool = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                recent_model_reserve_slots=5,
+                capacity_probe_interval_cycles=0,
+                capacity_state_path=state_path,
+            )
+            self.assertEqual(100, pool.available_submit_slots())
+            with self.assertRaises(ModelHubAPIError):
+                pool.add_task({"model": "rejected-during-race"})
+
+            self.assertEqual([100], pool.account_capacity_limits())
+            client.reject_capacity = False
+            pool.refresh_active_counts()
+            self.assertEqual(95, pool.old_model_submit_slots())
+            pool.add_task_for_model(
+                {"model": "eligible-old-model"},
+                model_last_modified="2025-01-01T00:00:00Z",
+                submitted_at="2026-10-02T00:00:00Z",
+            )
+            self.assertEqual(1, len(client.submitted))
+            restarted = ModelHubClientPool(
+                [client],  # type: ignore[list-item]
+                active_task_cap=100,
+                capacity_state_path=state_path,
+            )
+            self.assertEqual([100], restarted.account_capacity_limits())
+
     def test_exact_active_count_uses_filtered_platform_totals(self) -> None:
         client = ModelHubClient(token="test-token")
         calls: list[str] = []
@@ -481,8 +551,37 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
             recent_model_days=7,
         )
 
-        self.assertEqual(["configured_window", "last_7_days"], [stage["name"] for stage in stages])
+        self.assertEqual(
+            ["configured_window", "last_7_days", "recent_window_full"],
+            [stage["name"] for stage in stages],
+        )
         self.assertTrue(all(stage["updatedAfter"] >= now - timedelta(days=7) for stage in stages))
+        self.assertEqual(3000, stages[-1]["limit"])
+
+    def test_recent_only_scan_deepens_even_when_date_window_cannot_expand(self) -> None:
+        now = datetime(2026, 8, 11, tzinfo=timezone.utc)
+        for initial_cutoff in (None, now - timedelta(days=7), now - timedelta(days=30)):
+            with self.subTest(initial_cutoff=initial_cutoff):
+                stages = build_adaptive_scan_stages(
+                    now=now,
+                    initial_updated_after=initial_cutoff,
+                    initial_limit=20,
+                    max_models=800,
+                    allow_older_than_recent_window=False,
+                )
+                self.assertEqual([20, 800], [stage["limit"] for stage in stages])
+                self.assertTrue(all(stage["updatedAfter"] == now - timedelta(days=7) for stage in stages))
+
+    def test_recent_only_scan_deduplicates_an_already_full_window(self) -> None:
+        now = datetime(2026, 8, 11, tzinfo=timezone.utc)
+        stages = build_adaptive_scan_stages(
+            now=now,
+            initial_updated_after=now - timedelta(days=7),
+            initial_limit=100,
+            max_models=100,
+            allow_older_than_recent_window=False,
+        )
+        self.assertEqual(1, len(stages))
 
     def test_submit_candidate_reports_old_model_policy_skip_at_dynamic_threshold(self) -> None:
         client = FakeClient(active_count=95)
@@ -738,6 +837,54 @@ class ClientPoolConcurrencyTests(unittest.TestCase):
             self.assertEqual("last_modified", summary["scanStages"][-1]["sort"])
             self.assertIsNone(discovery.calls[-1])
 
+    def test_reserved_slots_find_recent_candidates_beyond_first_scan_prefix(self) -> None:
+        class RecentOnlyClient(ExhaustedRecentClient):
+            def old_model_submit_slots(self) -> int:
+                return 0
+
+        class PagedRecentDiscovery(FakeDiscovery):
+            def __init__(self) -> None:
+                super().__init__(4)
+                self.models[0] = replace(self.models[0], repo_id="owner/recent-exhausted")
+                self.limits: list[int] = []
+
+            def list_recent_models(self, *, limit, **_kwargs):  # noqa: ANN003
+                self.limits.append(limit)
+                return self.models[:limit]
+
+        with tempfile.TemporaryDirectory() as temporary_dir:
+            root = Path(temporary_dir)
+            args = build_parser().parse_args([
+                "--task-types", "text-generation", "--since-hours", "168",
+                "--limit", "1", "--max-scan-models", "4",
+                "--skip-outcome-sync", "--skip-history-archive",
+                "--disable-gpu-strategy",
+            ])
+            for field, filename in (
+                ("runs_dir", "runs"), ("ledger_path", "ledger.jsonl"),
+                ("outcomes_path", "outcomes.jsonl"), ("claims_path", "claims.jsonl"),
+                ("history_archive_path", "history.jsonl"),
+                ("submission_exclusions_path", "exclusions.jsonl"),
+            ):
+                setattr(args, field, str(root / filename))
+            selector = TemplateSelector()
+            discovery = PagedRecentDiscovery()
+            client = RecentOnlyClient(
+                available=2,
+                processed_gpus=set(selector.supported_target_gpus("text-generation", auto_only=True)),
+            )
+            summary = run_submission(
+                args,
+                now=datetime(2026, 1, 2, 12, tzinfo=timezone.utc),
+                hf_discovery=discovery,  # type: ignore[arg-type]
+                modelhub_client=client,  # type: ignore[arg-type]
+                template_selector=selector,
+            )
+            self.assertEqual([1, 4], discovery.limits)
+            self.assertEqual(2, summary["submittedCount"])
+            self.assertEqual("recent_window_full", summary["scanStages"][-1]["name"])
+            self.assertEqual(4, summary["scannedModels"])
+
     def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None:
         with tempfile.TemporaryDirectory() as temporary_dir:
             root = Path(temporary_dir)