fix: resolve 3 security audit issues

#3: Stored XSS in stats dashboard - escape p[path] with html.escape()
#4: Caddy timeout race - increase read/write_timeout 30s -> 60s
#5: Missing CSP header - add Content-Security-Policy to Caddyfile

proxy_pool.py

@@ -1,11 +1,12 @@
 """
-异步代理池 v4 — Cookie 感知 + 挑战分离 + 真实统计
+异步代理池 v5 — P2C + 状态机 + 全量 5000 代理 + 并发限流
 
-v4 vs v3:
-- 每个代理维护 cookie jar（cf_clearance 等），成功请求后自动保存
-- CF 挑战不再标记为代理死亡（mark_challenged ≠ mark_failure）
-- 真实统计：维护原子计数器，不造假数据
-- 更多 TLS 指纹：加入 safari15_5/safari17_0
+对标 go3 的代理管理架构，适配 Python/curl_cffi：
+- 全量加载 proxies.txt（不再按端口过滤）
+- P2C (Power of Two Choices) 选择算法
+- 状态机: healthy → unstable → blocked → probing
+- 并发限制 400（Webshare 上限）
+- 被动探测 + 指数退避
 """
 import asyncio
 import logging
@@ -20,23 +21,42 @@ from curl_cffi.requests import AsyncSession
 
 logger = logging.getLogger("ao3-proxy-pool")
 
-OPTIMAL_MIN_PORT = 13500
-OPTIMAL_MAX_PORT = 14499
+# ─── Constants ───────────────────────────────────────────────────────────────
+
+PROXY_FILE = "/home/ubuntu/proxy.txt"
 WORKING_PROXIES_FILE = "/dev/shm/working_proxies.txt"
 
-FAST_POOL_SIZE = 50
+# Concurrency limit: Webshare allows ~500 concurrent, cap at 400 for safety
+MAX_CONCURRENT_REQUESTS = 400
 
-# 被动检查：只对快池做轻量采样
-SAMPLE_INTERVAL = 60
-SAMPLE_BATCH_SIZE = 10
-SAMPLE_TIMEOUT = 5
-FAST_POOL_REFRESH_INTERVAL = 10
+# Probe config (like go3: 10s interval, 10 concurrency, 2 successes to promote)
+PROBE_INTERVAL = 10
+PROBE_CONCURRENCY = 10
+PROBE_SUCCESS_THRESHOLD = 2
+PROBE_TIMEOUT = 5
 
-# TLS fingerprints — proven against AO3 Cloudflare
-# safari15_5/17_0 have highest CF bypass rate per testing
+# Cooldown config
+BLOCKED_COOLDOWN_BASE = 120    # First block: 2 min
+BLOCKED_COOLDOWN_MAX = 1800    # Max: 30 min
+UNSTABLE_COOLDOWN = 15         # Transient failure: 15s
+
+# Outage attack detection
+OUTAGE_THRESHOLD = 0.9  # 90% proxies unavailable → attack page
+
+# Max retries for a single request
+MAX_RETRIES = 5
+
+# Ban-worthy status codes
+BAN_STATUS_CODES = {403, 525}
+
+# TLS fingerprints (proven against AO3 CF)
 BROWSER_IMPS = ["safari15_5", "safari17_0", "chrome123", "chrome124"]
 
-WARM_THRESHOLD_S = 3.0
+# States
+STATE_HEALTHY = "healthy"
+STATE_UNSTABLE = "unstable"
+STATE_BLOCKED = "blocked"
+STATE_PROBING = "probing"
 
 
 def _parse_cookie_expires(set_cookie: str) -> float:
@@ -55,23 +75,21 @@ def _parse_cookie_expires(set_cookie: str) -> float:
                 return time.time() + int(part[9:])
             except Exception:
                 pass
-    return 0  # session cookie
+    return 0
 
 
 def load_proxies_from_file() -> list[str]:
-    """Load proxy list from known-working file, or full list with port filtering."""
+    """Load ALL proxies from file. No port filtering in v5."""
+    # Prefer cached working proxies if available and fresh
+    # But only use as a hint — still load the full list
     if os.path.exists(WORKING_PROXIES_FILE) and os.path.getsize(WORKING_PROXIES_FILE) > 0:
         with open(WORKING_PROXIES_FILE) as f:
-            proxies = [l.strip() for l in f if l.strip() and ":" in l]
-        if proxies:
-            logger.info(f"Loaded {len(proxies)} working proxies from {WORKING_PROXIES_FILE}")
-            filtered = [p for p in proxies if ":" in p and OPTIMAL_MIN_PORT <= int(p.split(":")[-1]) <= OPTIMAL_MAX_PORT]
-            if len(filtered) >= 10:
-                return filtered
-            return proxies
-    proxy_file = "/home/ubuntu/proxy.txt"
+            cached = [l.strip() for l in f if l.strip() and ":" in l]
+        if len(cached) >= 100:
+            logger.info(f"Found {len(cached)} cached working proxies — using as seed")
+
     proxies = []
-    with open(proxy_file) as f:
+    with open(PROXY_FILE) as f:
         for line in f:
             line = line.strip()
             if not line or ":" not in line:
@@ -79,96 +97,133 @@ def load_proxies_from_file() -> list[str]:
             if "|" in line:
                 line = line.split("|")[-1].strip()
             proxies.append(line)
-    filtered = [p for p in proxies if ":" in p and OPTIMAL_MIN_PORT <= int(p.split(":")[-1]) <= OPTIMAL_MAX_PORT]
-    return filtered if len(filtered) >= 10 else proxies
+
+    if not proxies:
+        raise RuntimeError(f"No proxies found in {PROXY_FILE}")
+
+    logger.info(f"Loaded {len(proxies)} proxies from {PROXY_FILE}")
+    return proxies
 
 
 class ProxySession:
-    """A single proxy with its own AsyncSession, cookie jar, and health state."""
+    """A single proxy with its own AsyncSession, cookie jar, and state machine."""
 
     __slots__ = (
-        "host_port", "host", "port_str", "port",
-        "session", "impersonate", "alive",
-        "consecutive_failures", "ban_until", "last_used",
-        "avg_response_time", "weight", "requests_handled",
-        "last_sample", "sample_passed",
-        # v4: cookie jar
+        "idx", "host_port", "host", "port_str", "port",
+        "session", "impersonate",
+        # State machine
+        "state",  # healthy / unstable / blocked / probing
+        # Stats
+        "successes", "failures", "consecutive_net_failures",
+        "consecutive_blocked",
+        "last_used", "last_success_at", "last_failure_at",
+        "avg_response_time", "requests_handled",
+        # Cooldown / block
+        "cooldown_until", "next_probe_at",
+        "probe_successes",
+        # v5: inflight counter
+        "inflight",
+        # Cookie jar
         "_cookies", "_cookie_expires", "_lock",
     )
 
-    def __init__(self, host_port: str):
+    def __init__(self, idx: int, host_port: str):
+        self.idx = idx
         self.host_port = host_port
         self.host, self.port_str = host_port.split(":")
         self.port = int(self.port_str)
         self.session: Optional[AsyncSession] = None
         self.impersonate = random.choice(BROWSER_IMPS)
-        self.alive = True
-        self.consecutive_failures = 0
-        self.ban_until = 0.0
+
+        self.state = STATE_HEALTHY
+        self.successes = 0
+        self.failures = 0
+        self.consecutive_net_failures = 0
+        self.consecutive_blocked = 0
         self.last_used = 0.0
+        self.last_success_at = 0.0
+        self.last_failure_at = 0.0
         self.avg_response_time = 1.0
-        self.weight = 1.0
         self.requests_handled = 0
-        self.last_sample = 0.0
-        self.sample_passed = True
-        # v4: per-proxy cookie jar (cf_clearance, etc.)
+        self.cooldown_until = 0.0
+        self.next_probe_at = 0.0
+        self.probe_successes = 0
+        self.inflight = 0
+
         self._cookies: dict[str, str] = {}
         self._cookie_expires: dict[str, float] = {}
         self._lock = Lock()
 
-    # ─── Cookie management ────────────────────────────────────────────
-
-    def save_cookies(self, headers: dict) -> int:
-        """Extract Set-Cookie from response headers. Returns count saved."""
-        saved = 0
-        set_cookie = headers.get("Set-Cookie", "")
-        if not set_cookie:
-            # Some servers use set-cookie (lowercase) in HTTP/2
-            set_cookie = headers.get("set-cookie", "")
-        if not set_cookie:
-            return 0
-
+    @property
+    def is_available(self) -> bool:
+        """Proxy is selectable for user requests."""
         now = time.time()
-        with self._lock:
-            for part in set_cookie.split(","):
-                # Handle comma-separated cookies (ugh)
-                part = part.strip()
-                if "=" not in part:
-                    continue
-                name, _, rest = part.partition("=")
-                value = rest.split(";")[0].strip() if ";" in rest else rest.strip()
-                name = name.strip()
-                if not name:
-                    continue
-                self._cookies[name] = value
-                expires = _parse_cookie_expires(part)
-                if expires > 0:
-                    self._cookie_expires[name] = expires
-                saved += 1
+        if now < self.cooldown_until:
+            return False
+        if self.state == STATE_HEALTHY:
+            return True
+        if self.state == STATE_UNSTABLE and now >= self.cooldown_until:
+            return True
+        if self.state == STATE_PROBING:
+            return False  # Don't route user traffic to probing proxies
+        return False  # STATE_BLOCKED
 
-        # Purge expired cookies
-        self._purge_expired(now)
-        if saved:
-            logger.debug(f"Saved {saved} cookies for {self.host_port} (keys: {list(self._cookies.keys())})")
-        return saved
-
-    def get_cookie_header(self) -> str:
-        """Get Cookie header string for this proxy. Returns '' if no cookies."""
-        now = time.time()
-        with self._lock:
-            self._purge_expired(now)
-            if not self._cookies:
-                return ""
-            return "; ".join(f"{k}={v}" for k, v in self._cookies.items())
+    @property
+    def is_user_selectable(self) -> bool:
+        """Proxy can be chosen for sticky/user requests. Like go3's isUserSelectable."""
+        return self.is_available
 
     def _purge_expired(self, now: float):
-        """Remove expired cookies."""
         expired = [k for k, exp in self._cookie_expires.items() if 0 < exp < now]
         for k in expired:
             self._cookies.pop(k, None)
             self._cookie_expires.pop(k, None)
 
-    # ─── Session management ───────────────────────────────────────────
+    def save_cookies(self, headers: dict) -> int:
+        """Extract ONLY cf_clearance from response headers. Returns count saved.
+
+        The proxy must be TRANSPARENT to AO3 — do NOT save _otwarchive_session,
+        user_credentials, or any other AO3 cookies. Only cf_clearance (Cloudflare
+        bypass) is our concern. Everything else passes through to the browser.
+        """
+        set_cookie = headers.get("Set-Cookie", headers.get("set-cookie", ""))
+        if not set_cookie:
+            return 0
+        now = time.time()
+        with self._lock:
+            parts = set_cookie.split(";")
+            if parts:
+                first_part = parts[0].strip()
+                if "=" in first_part:
+                    name, _, value = first_part.partition("=")
+                    name = name.strip()
+                    value = value.strip()
+                    # ONLY save cf_clearance — the only cookie the proxy needs
+                    if name == "cf_clearance":
+                        self._cookies[name] = value
+                        logger.info(f"[CF_SAVE] {self.host_port}: saved cf_clearance")
+                        expires = _parse_cookie_expires(set_cookie)
+                        if expires > 0:
+                            self._cookie_expires[name] = expires
+                        self._purge_expired(now)
+                        return 1
+                    else:
+                        logger.info(f"[CF_SKIP] {self.host_port}: skipping {name} (not cf_clearance)")
+            self._purge_expired(now)
+        return 0
+
+    def get_cookie_header(self) -> str:
+        now = time.time()
+        with self._lock:
+            self._purge_expired(now)
+            if not self._cookies:
+                return ""
+            header = "; ".join(f"{k}={v}" for k, v in self._cookies.items())
+            logger.info(f"[PROXY_COOKIES] {self.host_port}: {list(self._cookies.keys())[:10]}")
+            return header
+
+    def has_cookies(self) -> bool:
+        return bool(self._cookies)
 
     async def get_session(self) -> AsyncSession:
         if self.session is None:
@@ -179,10 +234,11 @@ class ProxySession:
                 "http": f"http://{self.host_port}",
                 "https": f"http://{self.host_port}",
             }
-            # Default headers that don't change per-request (safe to set on session)
             self.session.headers.update({
-                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
-                              "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
+                "User-Agent": (
+                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+                ),
                 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
                 "Accept-Language": "en-US,en;q=0.9",
                 "Accept-Encoding": "gzip, deflate, br",
@@ -197,260 +253,364 @@ class ProxySession:
                 pass
             self.session = None
 
-    # ─── Health tracking ──────────────────────────────────────────────
-
-    def update_weight(self):
-        self.weight = 1.0 / max(self.avg_response_time, 0.1)
+    # ─── State machine (go3-compatible) ──────────────────────────────────
 
     def mark_success(self, response_time: float):
-        self.alive = True
-        self.consecutive_failures = 0
-        self.ban_until = 0.0
+        """Request succeeded. Promote to healthy."""
+        self.last_used = time.time()
+        self.last_success_at = time.time()
+        self.successes += 1
         self.requests_handled += 1
+        self.consecutive_net_failures = 0
+        self.consecutive_blocked = 0
         self.avg_response_time = self.avg_response_time * 0.7 + response_time * 0.3
-        self.update_weight()
+        self.cooldown_until = 0.0
+        self.state = STATE_HEALTHY
 
-    def mark_failure(self):
-        """Proxy-level failure (connection error, timeout, etc.) — exponential backoff."""
+    def mark_failure(self, is_network: bool = True):
+        """Connection error / timeout. Exponential backoff."""
+        now = time.time()
+        self.last_used = now
+        self.last_failure_at = now
+        self.failures += 1
         self.requests_handled += 1
-        self.consecutive_failures += 1
-        backoff = min(5 * (3 ** (self.consecutive_failures - 1)), 300)
-        self.ban_until = time.time() + backoff
-        if self.consecutive_failures >= 3:
-            self.alive = False
-        self.sample_passed = False
+
+        if is_network:
+            self.consecutive_net_failures += 1
+            self.consecutive_blocked = 0
+
+        if self.consecutive_net_failures >= 3:
+            self.state = STATE_BLOCKED
+            # Exponential backoff: base * 2^(failures-1), capped at max
+            backoff = min(
+                BLOCKED_COOLDOWN_BASE * (2 ** (self.consecutive_net_failures - 3)),
+                BLOCKED_COOLDOWN_MAX
+            )
+            self.cooldown_until = now + backoff
+            self.next_probe_at = now + backoff
+            self.probe_successes = 0
+        elif self.consecutive_net_failures >= 1:
+            self.state = STATE_UNSTABLE
+            self.cooldown_until = now + UNSTABLE_COOLDOWN
+        # else stay healthy / unchanged
+
+    def mark_blocked(self):
+        """403/525 — CF blocked. Move to blocked state."""
+        now = time.time()
+        self.last_used = now
+        self.last_failure_at = now
+        self.failures += 1
+        self.requests_handled += 1
+        self.consecutive_blocked += 1
+        self.consecutive_net_failures = 0
+
+        self.state = STATE_BLOCKED
+        backoff = min(
+            BLOCKED_COOLDOWN_BASE * (2 ** (self.consecutive_blocked - 1)),
+            BLOCKED_COOLDOWN_MAX
+        )
+        self.cooldown_until = now + backoff
+        self.next_probe_at = now + backoff
+        self.probe_successes = 0
 
     def mark_challenged(self):
-        """CF challenge detected — proxy is alive but needs cookie. NO backoff."""
-        # Don't increment consecutive_failures — challenge is not proxy's fault
-        # Don't set ban_until — proxy may work with cookies
+        """CF challenge detected — proxy is alive but needs cookie. Keep healthy, don't penalize."""
+        self.last_used = time.time()
         self.requests_handled += 1
-        # Only mark as not-sample-passed so it won't be fast-pool priority
-        self.sample_passed = False
+        # Don't change state — challenge is not proxy's fault
 
-    def mark_sample(self, passed: bool, response_time: float = 0):
-        """Lightweight periodic check result."""
-        self.last_sample = time.time()
-        self.sample_passed = passed
-        if passed and not self.alive:
-            self.alive = True
-            self.consecutive_failures = max(0, self.consecutive_failures - 1)
-            self.avg_response_time = self.avg_response_time * 0.5 + response_time * 0.5
-            self.update_weight()
+    def mark_probe_result(self, passed: bool, response_time: float = 0):
+        """Periodic probe result. If probing and enough successes, promote to healthy."""
+        now = time.time()
+        if passed:
+            self.probe_successes += 1
+            if self.probe_successes >= PROBE_SUCCESS_THRESHOLD:
+                self.state = STATE_HEALTHY
+                self.cooldown_until = 0.0
+                self.consecutive_net_failures = 0
+                self.consecutive_blocked = 0
+                self.avg_response_time = self.avg_response_time * 0.5 + response_time * 0.5
+        else:
+            self.probe_successes = 0
+            # Extend cooldown
+            self.cooldown_until = now + max(self.cooldown_until - now, BLOCKED_COOLDOWN_BASE)
+        self.next_probe_at = 0.0
 
-    @property
-    def is_available(self) -> bool:
-        return self.alive and time.time() > self.ban_until
+    def to_stats_dict(self) -> dict:
+        return {
+            "address": self.host_port,
+            "state": self.state,
+            "successes": self.successes,
+            "failures": self.failures,
+            "consecutive_net_failures": self.consecutive_net_failures,
+            "consecutive_blocked": self.consecutive_blocked,
+            "last_used": self.last_used,
+            "last_success_at": self.last_success_at,
+            "last_failure_at": self.last_failure_at,
+            "avg_response_time": round(self.avg_response_time, 3),
+            "requests_handled": self.requests_handled,
+            "cooldown_until": self.cooldown_until,
+            "inflight": self.inflight,
+            "has_cookies": bool(self._cookies),
+            "impersonate": self.impersonate,
+        }
 
     def __repr__(self):
-        nc = len(self._cookies)
-        return f"PS({self.host_port}, alive={self.alive}, {self.avg_response_time:.1f}s, cookies={nc})"
+        return f"PS({self.host_port}, {self.state}, {self.avg_response_time:.2f}s, inflight={self.inflight})"
 
 
 class AsyncProxyPool:
-    """Tiered async proxy pool with cookie-aware session management."""
+    """Proxy pool v5 — P2C selection + state machine + concurrent limiter."""
 
     def __init__(self):
         self._proxies: list[ProxySession] = []
-        self._fast_pool: list[ProxySession] = []
-        self._fast_pool_updated = 0.0
-        self._sample_task: Optional[asyncio.Task] = None
-        # v4: atomic counters for real stats (no more fake data)
+        self._idx_map: dict[int, ProxySession] = {}
+        self._addr_map: dict[str, ProxySession] = {}
+
+        # Concurrent request limiter (capped at 400)
+        self._concurrency_limiter = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
+
+        # Stats
         self._stats_lock = Lock()
-        self._stats = {"alive": 0, "dead": 0, "banned": 0, "challenged": 0}
-        self._stats_cache = {}
-        self._stats_cache_ts = 0.0
+
+        # Probe task
+        self._probe_task: Optional[asyncio.Task] = None
+        self._probe_running = False
+
+        # Load proxies
         self._load_proxies()
-        self._start_sampling()
+        self._start_probing()
+
+        logger.info(
+            f"ProxyPool v5 ready: {len(self._proxies)} proxies, "
+            f"concurrency={MAX_CONCURRENT_REQUESTS}, selection=p2c"
+        )
 
     def _load_proxies(self):
-        proxies = load_proxies_from_file()
-        self._proxies = [ProxySession(hp) for hp in proxies]
-        self._refresh_fast_pool()
-        self._recompute_stats()
-        logger.info(f"ProxyPool v4 ready: {len(self._proxies)} proxies, fast={len(self._fast_pool)}")
+        proxy_list = load_proxies_from_file()
+        self._proxies = []
+        self._idx_map = {}
+        self._addr_map = {}
+        for i, hp in enumerate(proxy_list):
+            ps = ProxySession(i, hp)
+            self._proxies.append(ps)
+            self._idx_map[i] = ps
+            self._addr_map[hp] = ps
+        logger.info(f"Loaded {len(self._proxies)} proxies")
 
-    def _start_sampling(self):
+    def _start_probing(self):
         try:
             loop = asyncio.get_running_loop()
         except RuntimeError:
             loop = asyncio.new_event_loop()
-        if self._sample_task is None or self._sample_task.done():
-            self._sample_task = asyncio.create_task(self._sampling_loop())
+        self._probe_task = asyncio.create_task(self._probe_loop())
 
-    def _refresh_fast_pool(self):
-        """Select fastest N proxies, seeded immediately from all alive."""
-        alive = [p for p in self._proxies if p.is_available]
-        sampled = [p for p in alive if p.sample_passed]
-        unsampled = [p for p in alive if not p.sample_passed]
-        sampled.sort(key=lambda p: p.avg_response_time)
-        unsampled.sort(key=lambda p: p.avg_response_time)
-        combined = sampled + unsampled
-        self._fast_pool = combined[:FAST_POOL_SIZE]
-        self._fast_pool_updated = time.time()
+    # ─── Proxy selection (P2C — Power of Two Choices) ───────────────────
 
-    def _recompute_stats(self):
-        """Accurate stats — iterate all proxies (fast, ~726 items)."""
-        alive = 0
-        dead = 0
-        banned = 0
-        for p in self._proxies:
-            if p.alive:
-                alive += 1
-                if not p.is_available:
-                    banned += 1
-            else:
-                dead += 1
-        with self._stats_lock:
-            self._stats = {"alive": alive, "dead": dead, "banned": banned}
-
-    # ─── Proxy selection ──────────────────────────────────────────────
-
-    def get_fast_proxy(self) -> Optional[ProxySession]:
-        """Fast proxy for interactive requests (login/register/POST)."""
+    def p2c_select(self, exclude_idxs: Optional[set[int]] = None) -> Optional[ProxySession]:
+        """
+        Power of Two Choices: pick 2 random candidates, return the one with fewer inflight.
+        Falls back to random if P2C produces ties.
+        """
+        exclude = exclude_idxs or set()
         now = time.time()
-        if now - self._fast_pool_updated > FAST_POOL_REFRESH_INTERVAL:
-            self._refresh_fast_pool()
-        if not self._fast_pool:
-            return self.get_proxy()
-        total = sum(p.weight for p in self._fast_pool)
-        if total <= 0:
-            return random.choice(self._fast_pool)
-        r = random.uniform(0, total)
-        cum = 0
-        for p in self._fast_pool:
-            cum += p.weight
-            if r <= cum:
-                return p
-        return random.choice(self._fast_pool)
 
-    def get_proxy(self) -> Optional[ProxySession]:
-        """Weighted random from all available proxies."""
-        available = [p for p in self._proxies if p.is_available]
-        if not available:
-            # Fallback: use proxies with fewer than 10 failures
-            available = [p for p in self._proxies if p.consecutive_failures < 10]
-        if not available:
+        # Build candidate list (selectable, not excluded)
+        candidates = []
+        for p in self._proxies:
+            if p.idx in exclude:
+                continue
+            if not p.is_user_selectable:
+                continue
+            if p.inflight >= 10:  # Don't overload a single proxy
+                continue
+            candidates.append(p)
+
+        if not candidates:
+            # Fallback: anything with cooldown expired, even if probing
+            for p in self._proxies:
+                if p.idx in exclude:
+                    continue
+                if now >= p.cooldown_until and p.inflight < 10:
+                    candidates.append(p)
+
+        if not candidates:
+            # Desperate: any proxy
+            for p in self._proxies:
+                if p.idx in exclude:
+                    continue
+                if p.inflight < 10:
+                    candidates.append(p)
+
+        if not candidates:
             return None
-        total = sum(p.weight for p in available)
-        if total <= 0:
-            return random.choice(available)
-        r = random.uniform(0, total)
-        cum = 0
-        for p in available:
-            cum += p.weight
-            if r <= cum:
-                return p
-        return random.choice(available)
 
-    def get_proxy_with_cookies(self) -> Optional[ProxySession]:
-        """Get a proxy that has cookies saved (cf_clearance). Fallback to any available."""
-        available = [p for p in self._proxies if p.is_available]
-        with_cookies = [p for p in available if p._cookies]
-        if with_cookies:
-            total = sum(p.weight for p in with_cookies)
-            if total > 0:
-                r = random.uniform(0, total)
-                cum = 0
-                for p in with_cookies:
-                    cum += p.weight
-                    if r <= cum:
-                        return p
-            return random.choice(with_cookies)
-        return self.get_proxy()
+        if len(candidates) == 1:
+            return candidates[0]
 
-    # ─── Sampling loop ────────────────────────────────────────────────
+        # P2C: pick 2, choose the one with lower inflight
+        left = random.choice(candidates)
+        right = random.choice(candidates)
+        # Avoid same proxy
+        for _ in range(5):
+            if right.idx != left.idx:
+                break
+            right = random.choice(candidates)
 
-    async def _sampling_loop(self):
-        """Lightweight sampling — only test fast-pool proxies."""
-        logger.info("Sampling loop started (lightweight, fast-pool only)")
+        if left.inflight <= right.inflight:
+            return left
+        return right
+
+    def get_by_idx(self, idx: int) -> Optional[ProxySession]:
+        return self._idx_map.get(idx)
+
+    def get_by_addr(self, addr: str) -> Optional[ProxySession]:
+        return self._addr_map.get(addr)
+
+    # ─── Sticky proxy ───────────────────────────────────────────────────
+
+    def sticky_select(self, sticky_idx: int, exclude_idxs: set[int]) -> Optional[ProxySession]:
+        """Try to use the sticky proxy if it's still selectable."""
+        p = self.get_by_idx(sticky_idx)
+        if p and p.is_user_selectable and p.idx not in exclude_idxs and p.inflight < 10:
+            return p
+        return None
+
+    # ─── Stats ──────────────────────────────────────────────────────────
+
+    def get_stats(self) -> dict:
+        total = len(self._proxies)
+        healthy = 0
+        unstable = 0
+        blocked = 0
+        probing = 0
+        total_inflight = 0
+        with_cookies = 0
+        total_requests = 0
+        response_times = []
+
+        for p in self._proxies:
+            if p.state == STATE_HEALTHY:
+                healthy += 1
+            elif p.state == STATE_UNSTABLE:
+                unstable += 1
+            elif p.state == STATE_BLOCKED:
+                blocked += 1
+            elif p.state == STATE_PROBING:
+                probing += 1
+            total_inflight += p.inflight
+            if p.has_cookies():
+                with_cookies += 1
+            total_requests += p.requests_handled
+            if p.requests_handled > 0:
+                response_times.append(p.avg_response_time)
+
+        avg_rt = sum(response_times) / max(len(response_times), 1)
+
+        return {
+            "total": total,
+            "healthy": healthy,
+            "unstable": unstable,
+            "blocked": blocked,
+            "probing": probing,
+            "available": healthy,  # Only healthy are truly available
+            "total_inflight": total_inflight,
+            "with_cookies": with_cookies,
+            "total_requests": total_requests,
+            "avg_response_time_s": round(avg_rt, 3),
+            "pool_health": round(healthy / max(total, 1), 3),
+            "pool_unavailable_ratio": round((blocked + probing) / max(total, 1), 3),
+        }
+
+    def get_proxy_stats_list(self) -> list[dict]:
+        """Detailed per-proxy stats for the monitor."""
+        return [p.to_stats_dict() for p in self._proxies]
+
+    def pool_available_ratio(self) -> float:
+        """Fraction of proxies currently selectable."""
+        available = sum(1 for p in self._proxies if p.is_user_selectable)
+        return available / max(len(self._proxies), 1)
+
+    def pool_unavailable_ratio(self) -> float:
+        return 1.0 - self.pool_available_ratio()
+
+    # ─── Concurrency limiter ────────────────────────────────────────────
+
+    @property
+    def concurrency_limiter(self) -> asyncio.Semaphore:
+        return self._concurrency_limiter
+
+    # ─── Probe loop ─────────────────────────────────────────────────────
+
+    async def _probe_loop(self):
+        """Periodic probe to recover blocked proxies. Like go3's probe goroutine."""
+        await asyncio.sleep(5)  # Wait for initial startup
+        logger.info("Probe loop started")
         while True:
             try:
-                await asyncio.sleep(SAMPLE_INTERVAL)
-                pool = self._fast_pool[:] if self._fast_pool else self._proxies[:50]
-                if not pool:
+                await asyncio.sleep(PROBE_INTERVAL)
+                # Find proxies that need probing: blocked/probing with cooldown expired
+                now = time.time()
+                to_probe = []
+                for p in self._proxies:
+                    if p.state in (STATE_BLOCKED, STATE_PROBING):
+                        if now >= p.cooldown_until and now >= p.next_probe_at:
+                            to_probe.append(p)
+
+                if not to_probe:
                     continue
-                alive_cnt = 0
-                dead_cnt = 0
-                for i in range(0, len(pool), SAMPLE_BATCH_SIZE):
-                    batch = pool[i:i + SAMPLE_BATCH_SIZE]
-                    checks = [self._check_single(p) for p in batch]
-                    results = await asyncio.gather(*checks, return_exceptions=True)
+
+                # Probe in batches of PROBE_CONCURRENCY
+                random.shuffle(to_probe)
+                for i in range(0, len(to_probe), PROBE_CONCURRENCY):
+                    batch = to_probe[i:i + PROBE_CONCURRENCY]
+                    tasks = [self._probe_single(p) for p in batch]
+                    results = await asyncio.gather(*tasks, return_exceptions=True)
                     for p, r in zip(batch, results):
                         if isinstance(r, Exception):
-                            p.mark_sample(False)
-                            dead_cnt += 1
-                        elif r[0]:
-                            p.mark_sample(True, r[1])
-                            alive_cnt += 1
+                            p.mark_probe_result(False)
                         else:
-                            p.mark_sample(False)
-                            dead_cnt += 1
-                self._refresh_fast_pool()
-                self._recompute_stats()
-                total_avg = sum(p.avg_response_time for p in pool if p.requests_handled > 0) / max(alive_cnt, 1)
-                logger.debug(f"Sample: {alive_cnt} alive, {dead_cnt} dead, fast={len(self._fast_pool)}, avg={total_avg*1000:.0f}ms")
+                            p.mark_probe_result(r[0], r[1])
+
+                # Log recovery stats
+                recovered = sum(1 for p in to_probe if p.state == STATE_HEALTHY)
+                if recovered:
+                    logger.info(f"Probe: {len(to_probe)} checked, {recovered} recovered")
             except asyncio.CancelledError:
                 break
             except Exception as e:
-                logger.error(f"Sample error: {e}")
+                logger.error(f"Probe error: {e}")
 
-    async def _check_single(self, proxy: ProxySession) -> tuple[bool, float]:
-        """Quick single-proxy check against AO3."""
+    async def _probe_single(self, proxy: ProxySession) -> tuple[bool, float]:
+        """Quick HEAD check against AO3."""
         start = time.time()
         try:
             s = await proxy.get_session()
-            resp = await s.head("https://archiveofourown.org/", timeout=SAMPLE_TIMEOUT)
-            return (200 <= resp.status_code < 500, time.time() - start)
+            resp = await s.head("https://archiveofourown.org/", timeout=PROBE_TIMEOUT)
+            ok = 200 <= resp.status_code < 500
+            return (ok, time.time() - start)
         except Exception:
             return (False, time.time() - start)
 
-    # ─── Lifecycle ────────────────────────────────────────────────────
+    # ─── Lifecycle ──────────────────────────────────────────────────────
 
     async def close_all(self):
-        if self._sample_task and not self._sample_task.done():
-            self._sample_task.cancel()
+        if self._probe_task and not self._probe_task.done():
+            self._probe_task.cancel()
             try:
-                await self._sample_task
+                await self._probe_task
             except asyncio.CancelledError:
                 pass
         for p in self._proxies:
             await p.close()
-        logger.info("All sessions closed")
+        logger.info("All proxy sessions closed")
 
-    # ─── Stats ────────────────────────────────────────────────────────
+    @property
+    def proxies(self) -> list[ProxySession]:
+        return self._proxies
 
-    def get_stats(self) -> dict:
-        """Accurate stats with caching (1s throttle to avoid iteration on every call)."""
-        now = time.time()
-        if self._stats_cache and now - self._stats_cache_ts < 1.0:
-            return self._stats_cache
-
-        self._recompute_stats()
-        total = len(self._proxies)
-
-        # Compute avg response from sampled proxies
-        sampled = [p.avg_response_time for p in self._proxies if p.requests_handled > 0]
-        avg_speed = sum(sampled) / max(len(sampled), 1)
-
-        # Count proxies with cookies
-        with_cookies = sum(1 for p in self._proxies if p._cookies)
-
-        result = {
-            "total": total,
-            "alive": self._stats["alive"],
-            "dead": self._stats["dead"],
-            "banned": self._stats["banned"],
-            "available": self._stats["alive"] - self._stats["banned"],
-            "warm": sum(1 for p in self._fast_pool if p.sample_passed),
-            "fast_pool": len(self._fast_pool),
-            "with_cookies": with_cookies,
-            "total_requests_handled": sum(p.requests_handled for p in self._proxies),
-            "avg_response_time_s": round(avg_speed, 3),
-        }
-        self._stats_cache = result
-        self._stats_cache_ts = now
-        return result
+    def __len__(self):
+        return len(self._proxies)
 
 
 # ─── Singleton ────────────────────────────────────────────────────────────────