fix(verify): distinguish bot challenges from dead sources

Author: Seungpyo1007

app/verify/cli.py

@@ -453,11 +453,22 @@ def cmd_check_urls(args: argparse.Namespace) -> int:
 def _summarize_cache(cache: dict[str, dict[str, Any]], targets: list[str]) -> None:
     from collections import Counter
     alive = sum(1 for u in targets if cache.get(u, {}).get("alive"))
-    dead = sum(1 for u in targets if u in cache and not cache[u].get("alive"))
-    print(f"\nliveness over {len(targets)} targeted URL(s): {alive} alive, {dead} dead")
+    indeterminate = sum(
+        1 for u in targets if http_check.is_automation_challenge(cache.get(u, {}))
+    )
+    dead = sum(
+        1 for u in targets
+        if u in cache and not cache[u].get("alive")
+        and not http_check.is_automation_challenge(cache[u])
+    )
+    print(
+        f"\nliveness over {len(targets)} targeted URL(s): {alive} alive, {dead} dead, "
+        f"{indeterminate} automation-challenged"
+    )
     reasons = Counter(
         cache[u].get("reason") for u in targets
         if u in cache and not cache[u].get("alive")
+        and not http_check.is_automation_challenge(cache[u])
     )
     if reasons:
         print("dead reasons:")
@@ -619,10 +630,17 @@ def cmd_pr(args: argparse.Namespace) -> int:
         except Exception as exc:  # network hiccup must not sink the report
             print(f"_Tier 1 skipped: {exc}_\n")
         alive = sum(1 for e in url_cache.values() if e.get("alive"))
-        dead = len(url_cache) - alive
+        indeterminate = sum(1 for e in url_cache.values() if http_check.is_automation_challenge(e))
+        dead = len(url_cache) - alive - indeterminate
         print("### Tier 1 — source-URL liveness (changed)\n")
-        print(f"Checked **{len(url_cache)}** unique URL(s): **{alive} alive**, **{dead} dead**.\n")
-        dead_reasons = Counter(e["reason"] for e in url_cache.values() if not e.get("alive"))
+        print(
+            f"Checked **{len(url_cache)}** unique URL(s): **{alive} alive**, "
+            f"**{dead} dead**, **{indeterminate} automation-challenged**.\n"
+        )
+        dead_reasons = Counter(
+            e["reason"] for e in url_cache.values()
+            if not e.get("alive") and not http_check.is_automation_challenge(e)
+        )
         if dead_reasons:
             print("| Dead reason | Count |")
             print("| --- | ---: |")

app/verify/http_check.py

@@ -36,6 +36,7 @@
 # nothing about whether the page exists, so these answers must never be cached as
 # a verdict — otherwise one impatient run marks a whole host dead for a TTL.
 TRANSIENT_STATUSES = frozenset({429, 503})
+AUTOMATION_CHALLENGE_REASON = "automation-challenge"
 RETRY_ATTEMPTS = 3
 RETRY_BACKOFF_S = (2.0, 6.0)
 MAX_RETRY_AFTER_S = 15.0
@@ -58,6 +59,11 @@ class CheckResult(NamedTuple):
     def transient(self) -> bool:
         return self.status in TRANSIENT_STATUSES
 
+    @property
+    def indeterminate(self) -> bool:
+        """True when the endpoint reached an anti-bot challenge, not a page verdict."""
+        return self.reason == AUTOMATION_CHALLENGE_REASON
+
 
 # --- opener abstraction (injectable for tests) -----------------------------------
 
@@ -102,7 +108,15 @@ def _is_homepage_redirect(original: str, final: str) -> bool:
     return _path_depth(original) >= 1 and _path_depth(final) == 0
 
 
-def classify(original_url: str, status: int | None, final_url: str | None) -> tuple[bool, str]:
+def classify(
+    original_url: str,
+    status: int | None,
+    final_url: str | None,
+    *,
+    automation_challenge: bool = False,
+) -> tuple[bool, str]:
+    if automation_challenge:
+        return False, AUTOMATION_CHALLENGE_REASON
     if status is None:
         return False, "error"
     if status >= 400:
@@ -122,11 +136,34 @@ def _retry_after_seconds(exc: Exception) -> float | None:
         return None  # HTTP-date form; fall back to our own backoff
 
 
-def _attempt(url: str, opener: Any) -> tuple[int | None, str | None, float | None]:
-    """One HEAD-then-GET pass. Returns (status, final_url, retry_after)."""
+def _is_automation_challenge(exc: Exception, code: int) -> bool:
+    """Recognize an explicit anti-bot challenge without weakening generic 403s.
+
+    A September 2026 probe with this module's User-Agent found
+    ``browser.geekbench.com`` returning ``403`` with
+    ``CF-Mitigated: challenge`` and Cloudflare's challenge page.  That proves the
+    request reached the cited host but not whether an unauthenticated client may
+    inspect the page.  Keep ordinary 401/403 responses as dead: an HTTP status
+    alone cannot distinguish a missing or forbidden page from bot protection.
+    """
+    if code != 403:
+        return False
+    headers = getattr(exc, "headers", None)
+    mitigated = headers.get("CF-Mitigated", "") if headers is not None else ""
+    return isinstance(mitigated, str) and mitigated.lower() == "challenge"
+
+
+def _attempt(url: str, opener: Any) -> tuple[int | None, str | None, float | None, bool]:
+    """One HEAD-then-GET pass.
+
+    Returns ``(status, final_url, retry_after, automation_challenge)``.  The last
+    value is intentionally based on an explicit gateway signal, never on a bare
+    403, so unavailable pages cannot become promotable merely by returning 403.
+    """
     status: int | None = None
     final: str | None = None
     retry_after: float | None = None
+    automation_challenge = False
     for method in ("HEAD", "GET"):
         try:
             status, final = opener.open(url, method)
@@ -138,11 +175,12 @@ def _attempt(url: str, opener: Any) -> tuple[int | None, str | None, float | Non
             if isinstance(code, int):
                 status, final = code, getattr(exc, "url", None) or url
                 retry_after = _retry_after_seconds(exc)
+                automation_challenge = _is_automation_challenge(exc, code)
                 if method == "HEAD" and code in (400, 403, 405, 501):
                     continue
                 break
-            status, final = None, None
-    return status, final, retry_after
+            status, final, automation_challenge = None, None, False
+    return status, final, retry_after, automation_challenge
 
 
 def check_one(
@@ -154,15 +192,18 @@ def check_one(
     to wait, not that the page is gone.
     """
     status = final = retry_after = None
+    automation_challenge = False
     for attempt in range(RETRY_ATTEMPTS):
-        status, final, retry_after = _attempt(url, opener)
+        status, final, retry_after, automation_challenge = _attempt(url, opener)
         if status not in TRANSIENT_STATUSES:
             break
         if on_rate_limit is not None:
             on_rate_limit(host_of(url))
         if attempt < RETRY_ATTEMPTS - 1:
             time.sleep(retry_after if retry_after is not None else RETRY_BACKOFF_S[attempt])
-    alive, reason = classify(url, status, final)
+    alive, reason = classify(
+        url, status, final, automation_challenge=automation_challenge
+    )
     return CheckResult(url, status, final, alive, reason)
 
 
@@ -247,13 +288,20 @@ def _task(url: str) -> CheckResult:
 # --- cache -----------------------------------------------------------------------
 
 
+def is_automation_challenge(entry: dict[str, Any]) -> bool:
+    """Whether a cached result reached an explicit anti-bot challenge."""
+    return entry.get("reason") == AUTOMATION_CHALLENGE_REASON
+
+
 def load_cache(path: Path = URL_CACHE_PATH) -> dict[str, dict[str, Any]]:
     """Load the cache, dropping rate-limit answers written by older runs.
 
     A 429/503 is not a verdict, so an entry holding one is not a cache hit —
     it is a URL we still have to check. Filtering on load heals a cache that a
     previous run poisoned (3,998 GSMArena pages were parked as dead this way,
-    all of which answer 200 when asked at a civil pace).
+    all of which answer 200 when asked at a civil pace). Explicit anti-bot
+    challenges stay cached as *indeterminate*: the promotion layer can use that
+    reachability signal only for an already-authoritative cited host.
     """
     return {
         e["url"]: e
@@ -292,14 +340,14 @@ def result_to_entry(r: CheckResult, ts: str) -> dict[str, Any]:
 
 
 def record_liveness(source_urls: list[str], cache: dict[str, dict[str, Any]]) -> tuple[int, int]:
-    """(#live, #dead) for a record's URLs that are present in the cache."""
+    """(#live, #dead) for cached URLs, excluding anti-bot indeterminate results."""
     live = dead = 0
     for u in source_urls:
         e = cache.get(u)
         if e is None:
             continue
         if e.get("alive"):
             live += 1
-        else:
+        elif not is_automation_challenge(e):
             dead += 1
     return live, dead

app/verify/promote.py

@@ -19,7 +19,7 @@
 from pathlib import Path
 from typing import Any, NamedTuple
 
-from . import hosts
+from . import hosts, http_check
 from .common import STATE_DIR
 
 CROSSREF_CACHE_PATH = STATE_DIR / "crossref_cache.jsonl"
@@ -36,16 +36,27 @@ class PromotionDecision(NamedTuple):
 def has_live_authoritative_source(
     source_urls: list[str], url_cache: dict[str, dict[str, Any]]
 ) -> bool:
-    """True if some cited URL is an authoritative host (Tier 1 *or* Tier 2) AND
-    confirmed alive. The green band already requires a T1/T2 source + completeness
-    + consistency; this just adds "and that source actually resolves". Requiring a
-    *manufacturer/encyclopaedia* (T1 only) was too strict — it never promoted the
-    many green records sourced from reputable spec/benchmark DBs (gsmarena,
-    cpubenchmark, ...), so verified never moved off its floor.
+    """True if an authoritative cited URL has a usable liveness signal.
+
+    A normal alive response always qualifies. An explicit anti-bot challenge also
+    qualifies, but only for an already classified Tier-1/Tier-2 host: it proves
+    the request reached that host while the gateway withheld page inspection.
+    This preserves the live-source gate without treating every 403 as a live
+    record. Generic errors, ordinary 401/403, redirects to a homepage, and
+    unclassified hosts remain insufficient for promotion.
+
+    This distinction is deliberate. The September 2026 probe found Geekbench's
+    Cloudflare challenge under the verifier User-Agent, while several other
+    Tier-2 databases returned normal 200 responses. Treating challenge responses
+    as dead made a host-wide automation policy look like thousands of dead
+    citations; dropping the liveness gate altogether made unchecked citations
+    promotable. This narrow fallback avoids both failure modes.
     """
     for u in source_urls:
         entry = url_cache.get(u)
-        if entry and entry.get("alive") and hosts.tier_of_host(hosts.host_of(u)) in (1, 2):
+        if not entry or hosts.tier_of_host(hosts.host_of(u)) not in (1, 2):
+            continue
+        if entry.get("alive") or http_check.is_automation_challenge(entry):
             return True
     return False
 

tests/verify/test_http_check.py

@@ -92,6 +92,14 @@ def test_record_liveness():
     assert http_check.record_liveness(["a", "b", "c", "missing"], cache) == (2, 1)
 
 
+def test_record_liveness_excludes_automation_challenge():
+    cache = {
+        "blocked": {"alive": False, "reason": http_check.AUTOMATION_CHALLENGE_REASON},
+        "dead": {"alive": False, "reason": "http-404"},
+    }
+    assert http_check.record_liveness(["blocked", "dead"], cache) == (0, 1)
+
+
 def test_cache_roundtrip():
     # tmp_path fixture is unreliable on this Windows runner; use a local scratch file.
     from pathlib import Path
@@ -115,6 +123,16 @@ def __init__(self, url, retry_after=None):
         self.headers = {"Retry-After": retry_after} if retry_after else {}
 
 
+class _Http403Challenge(Exception):
+    """urllib-shaped Cloudflare bot challenge observed from Geekbench."""
+
+    def __init__(self, url):
+        super().__init__("Forbidden")
+        self.code = 403
+        self.url = url
+        self.headers = {"CF-Mitigated": "challenge"}
+
+
 class FlakyOpener(FakeOpener):
     """Rate-limits the first ``fail_times`` calls, then answers normally."""
 
@@ -146,6 +164,27 @@ def test_persistent_rate_limit_is_transient_not_dead(monkeypatch):
     assert res.status == 429 and res.transient  # caller must not cache this as a verdict
 
 
+def test_cloudflare_bot_challenge_is_indeterminate_not_dead():
+    url = "https://browser.geekbench.com/v6/cpu/1"
+    op = FakeOpener({url: _Http403Challenge(url)})
+    res = http_check.check_one(url, op)
+    assert res.status == 403
+    assert not res.alive
+    assert res.indeterminate
+    assert not res.transient
+    assert res.reason == http_check.AUTOMATION_CHALLENGE_REASON
+    assert [method for _url, method in op.calls] == ["HEAD", "GET"]
+
+
+def test_generic_403_stays_dead():
+    err = type("E", (Exception,), {"code": 403, "url": None, "headers": {}})()
+    url = "https://example.com/private"
+    res = http_check.check_one(url, FakeOpener({url: err}))
+    assert not res.alive
+    assert not res.indeterminate
+    assert res.reason == "http-403"
+
+
 def test_rate_limit_slows_the_host_down(monkeypatch):
     monkeypatch.setattr(http_check.time, "sleep", lambda _s: None)
     url = "https://www.gsmarena.com/z-3.php"

tests/verify/test_promote_crossref.py

@@ -141,6 +141,40 @@ def test_green_does_not_promote_with_dead_authoritative_source():
     assert not d.promote
 
 
+def test_green_promotes_with_authoritative_automation_challenge():
+    url = "https://browser.geekbench.com/v6/cpu/1"
+    d = promote.decide(
+        band="green",
+        source_urls=[url],
+        url_cache={
+            url: {
+                "alive": False,
+                "status": 403,
+                "reason": "automation-challenge",
+            }
+        },
+        crossref_decision=None,
+    )
+    assert d.promote and d.reason == "green-live-source"
+
+
+def test_green_does_not_promote_with_unclassified_automation_challenge():
+    url = "https://example.com/specs/1"
+    d = promote.decide(
+        band="green",
+        source_urls=[url],
+        url_cache={
+            url: {
+                "alive": False,
+                "status": 403,
+                "reason": "automation-challenge",
+            }
+        },
+        crossref_decision=None,
+    )
+    assert not d.promote
+
+
 def test_yellow_without_confirm_holds():
     d = promote.decide(band="yellow", source_urls=[], url_cache={}, crossref_decision=None)
     assert not d.promote