fix: prevent stale capacity poisoning and deepen recent-only refill scans

Keep ordinary capacity races from lowering persistent per-account limits.
Deepen recent-only discovery within the existing age and quality gates.
Add offline regressions for recovery and reserved-slot refill.

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,38 @@ def make_candidate(index: int) -> dict:
 
 
 class ClientPoolConcurrencyTests(unittest.TestCase):
+    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 +514,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 +800,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)