From b3a4449856f555e31d9d5bb05e2c889ba5601d48 Mon Sep 17 00:00:00 2001
From: zydo <zydo.git@gmail.com>
Date: Thu, 18 Jun 2026 19:55:14 +0000
Subject: [PATCH] Surface engine cooldown and classify network-level blocks on
 /monitor

The /monitor page showed Yahoo as "healthy" while it was being blocked,
and gave no indication of which engines the scraper had benched. Two
gaps, both fixed here.

Recognize network-level blocks: classify_engine and the scraper's
cooldown tracker only treated HTTP 429/403/503 as a block, but Yahoo
blocks at the TCP layer (net::ERR_CONNECTION_CLOSED) with no HTTP
status, so it slipped through as healthy/degraded and was never benched.
A new common/block_signals.is_network_block recognizes connection-level
teardowns (not plain timeouts); it feeds both the cooldown tracker (Yahoo
now gets benched instead of eating a nav timeout every cycle) and
classify_engine (labelled "blocked" despite a NULL status). The metrics
aggregate now carries last_error_message so the API can see it.

Surface cooldown state: the EngineCooldownTracker lives in the scraper
process and the API couldn't see it, so there was no "cooling down"
concept at all. The scraper now snapshots tracker state to a new
engine_cooldowns table each cycle (next_probe_at as absolute UTC, since
the monotonic clock can't cross processes); /metrics overlays a
"cooldown" health label with a probe countdown and synthesizes rows for
benched engines that produced no logs, so they stay visible instead of
vanishing. A stale snapshot (scraper down) is ignored.

Docs (README, OBSERVABILITY, API_REFERENCE, BLOCK_SIGNAL_*) updated.
---
 README.md                                 |   4 +-
 api/static/monitor.html                   |   2 +-
 api/static/monitor.js                     |  19 ++++
 api/static/styles.css                     |   7 ++
 api/v1/metrics.py                         | 105 ++++++++++++++++++++--
 common/block_signals.py                   |  42 +++++++++
 common/database.py                        |  69 +++++++++++++-
 docs/API_REFERENCE.md                     |   6 +-
 docs/BLOCK_SIGNAL_CHARACTERIZATION.md     |   7 +-
 docs/BLOCK_SIGNAL_FINDINGS.md             |  17 ++--
 docs/OBSERVABILITY.md                     |  15 +++-
 postgres/init.sql                         |  14 +++
 scraper/cooldown.py                       |  34 ++++++-
 scraper/main.py                           |  14 +++
 tests/integration/conftest.py             |   4 +-
 tests/integration/test_api_integration.py |  35 ++++++++
 tests/test_engine_cooldown.py             |  37 ++++++++
 tests/test_metrics.py                     |  79 +++++++++++++++-
 18 files changed, 480 insertions(+), 30 deletions(-)
 create mode 100644 common/block_signals.py

diff --git a/README.md b/README.md
index 9e1a224..388c761 100644
--- a/README.md
+++ b/README.md
@@ -183,10 +183,10 @@ http://localhost:5000
 A built-in observability console at **`/monitor`** (linked from the wire masthead) gives a per-engine, per-cycle view of scraper health — no Prometheus/Grafana stack required, since the data already lives in Postgres. It polls [`GET /api/v1/metrics`](docs/API_REFERENCE.md#metrics) over a selectable window (1h / 6h / 24h) and shows:
 
 - **Overall strip** - active topics, total filed, scrape success rate, feed freshness, last-cycle duration, and scrapes-in-window (blocked / failed).
-- **Engines table** - one row per engine with a health label (`healthy` / `degraded` / `blocked` / `parsing` / `idle`), success %, fetch latency (avg / p95), items parsed, 0-parse count (selector-rot signal), blocks (429/403/503), and last HTTP status.
+- **Engines table** - one row per engine with a health label (`healthy` / `degraded` / `blocked` / `parsing` / `cooldown` / `idle`), success %, fetch latency (avg / p95), items parsed, 0-parse count (selector-rot signal), blocks (429/403/503), and last HTTP status. `blocked` now also covers connection-level teardowns with no HTTP status (e.g. Yahoo's `ERR_CONNECTION_CLOSED`).
 - **Recent cycles** - a sparkline of per-cycle durations plus a list (duration, topics, parsed, new events).
 
-A throttled engine shows up here rather than silently vanishing from the feed — below, Brave reads `degraded` with its block count while Google/Bing/Yahoo stay `healthy`. See [Observability](docs/OBSERVABILITY.md) for details.
+A throttled engine shows up here rather than silently vanishing from the feed. When the scraper benches an engine, the table shows `cooldown` with a countdown to the next probe — so an engine that produces no scrapes while cooling stays visible instead of disappearing. See [Observability](docs/OBSERVABILITY.md) for details.
 
 <p align="center">
 <img src="docs/pic/ui_monitor_screenshot.png" alt="TopicStreams /monitor ops page - per-engine health table and recent-cycle timeline" width="600"/>
diff --git a/api/static/monitor.html b/api/static/monitor.html
index 9866cb8..492c9e8 100644
--- a/api/static/monitor.html
+++ b/api/static/monitor.html
@@ -103,7 +103,7 @@ <h2>Recent failures</h2>
     <template id="engine-row-template">
         <tr>
             <td class="c-engine"><span class="edot"></span><span class="ename"></span></td>
-            <td class="c-health"><span class="ehealth"></span></td>
+            <td class="c-health"><span class="ehealth"></span><span class="ecooldown"></span></td>
             <td class="c-scrapes num r"></td>
             <td class="c-success num r"></td>
             <td class="c-latency num r"></td>
diff --git a/api/static/monitor.js b/api/static/monitor.js
index 2e5e826..6a28de0 100644
--- a/api/static/monitor.js
+++ b/api/static/monitor.js
@@ -110,6 +110,18 @@ class MonitorApp {
             hl.dataset.health = e.health;
             row.querySelector('.c-health').dataset.health = e.health;
 
+            // Cooldown: show the countdown to the next probe beside the label.
+            const cd = row.querySelector('.ecooldown');
+            if (e.cooldown_seconds_remaining != null) {
+                cd.textContent = `~${this.fmtCountdown(e.cooldown_seconds_remaining)}`;
+                const blocks = e.cooldown_failures
+                    ? `${e.cooldown_failures} consecutive block${e.cooldown_failures === 1 ? '' : 's'}; `
+                    : '';
+                cd.title = `${blocks}probing again in ~${this.fmtCountdown(e.cooldown_seconds_remaining)}`;
+            } else {
+                cd.textContent = '';
+            }
+
             row.querySelector('.c-scrapes').textContent = this.fmtInt(e.scrapes);
             row.querySelector('.c-success').textContent = this.fmtPct(e.success_rate);
             row.querySelector('.c-latency').textContent = `${this.fmtMs(e.avg_latency_ms)} / ${this.fmtMs(e.p95_latency_ms)}`;
@@ -233,6 +245,13 @@ class MonitorApp {
         return ms >= 1000 ? `${(ms / 1000).toFixed(1)}s` : `${ms}ms`;
     }
 
+    fmtCountdown(s) {
+        if (s == null) return '–';
+        if (s < 90) return `${Math.round(s)}s`;
+        if (s < 3600) return `${Math.round(s / 60)}m`;
+        return `${(s / 3600).toFixed(1)}h`;
+    }
+
     fmtFresh(s) {
         if (s == null) return '–';
         if (s < 60) return `${Math.round(s)}s`;
diff --git a/api/static/styles.css b/api/static/styles.css
index 3531592..6cb2b16 100644
--- a/api/static/styles.css
+++ b/api/static/styles.css
@@ -507,14 +507,21 @@ a { color: inherit; }
 .edot[data-health="healthy"] { background: var(--good); }
 .edot[data-health="degraded"] { background: var(--warn); }
 .edot[data-health="blocked"], .edot[data-health="parsing"] { background: var(--signal); }
+.edot[data-health="cooldown"] { background: var(--warn); }
 .edot[data-health="idle"] { background: var(--ink-faint); }
 
 .ehealth { font-size: 10.5px; letter-spacing: 0.08em; text-transform: uppercase; }
 .ehealth[data-health="healthy"] { color: var(--good); }
 .ehealth[data-health="degraded"] { color: var(--warn); }
 .ehealth[data-health="blocked"], .ehealth[data-health="parsing"] { color: var(--signal); }
+.ehealth[data-health="cooldown"] { color: var(--warn); }
 .ehealth[data-health="idle"] { color: var(--ink-faint); }
 
+/* Probe countdown shown beside a "cooldown" label, e.g. "~286s". Not uppercased
+   (it's data, not a status word) and de-emphasized vs. the label. */
+.ecooldown { margin-left: 6px; font-size: 10.5px; color: var(--ink-soft);
+  font-variant-numeric: tabular-nums; }
+
 td.hot { color: var(--signal); font-weight: 600; }
 td.warm { color: var(--warn); font-weight: 600; }
 
diff --git a/api/v1/metrics.py b/api/v1/metrics.py
index be77424..7af1109 100644
--- a/api/v1/metrics.py
+++ b/api/v1/metrics.py
@@ -14,6 +14,7 @@
 from starlette.concurrency import run_in_threadpool
 
 from common import database as db
+from common.block_signals import is_network_block
 
 router = APIRouter(prefix="/metrics", tags=["metrics"])
 
@@ -27,6 +28,12 @@
 # unavailable). Drives the per-engine "blocked" health label.
 _BLOCK_STATUSES = (429, 403, 503)
 
+# A persisted cooldown snapshot older than this is ignored: the scraper writes
+# one every cycle, so a stale snapshot means the scraper is down (the cycle
+# timeline surfaces that) and the cooldown countdown is meaningless. Generous
+# vs. the scrape interval so a slow cycle doesn't blink the indicator off.
+_COOLDOWN_STALE_SECONDS = 600
+
 
 def _naive_utc_now() -> datetime:
     return datetime.now(timezone.utc).replace(tzinfo=None)
@@ -40,7 +47,9 @@ def classify_engine(row: dict) -> str:
     """Per-engine health label from an aggregate row.
 
     idle    - no scrapes in the window
-    blocked - the most recent scrape was a throttle/block (429/403/503)
+    blocked - the most recent scrape was a throttle/block: a 429/403/503 status,
+              or a connection-level teardown with no HTTP status (e.g. Yahoo's
+              ERR_CONNECTION_CLOSED — see common/block_signals.py)
     parsing - sustained selector rot: every successful scrape parsed nothing
               (>=3 scrapes, >=1 success, all successes 0-entry)
     degraded- success rate below 0.75 (includes a total failure: 0%)
@@ -53,9 +62,9 @@ def classify_engine(row: dict) -> str:
     scrapes = row["scrapes"]
     if scrapes == 0:
         return "idle"
-    if (
-        row.get("last_success") is False
-        and row.get("last_http_status") in _BLOCK_STATUSES
+    if row.get("last_success") is False and (
+        row.get("last_http_status") in _BLOCK_STATUSES
+        or is_network_block(row.get("last_error_message"))
     ):
         return "blocked"
     successes = row["successes"]
@@ -65,6 +74,46 @@ def classify_engine(row: dict) -> str:
     return "degraded" if rate < 0.75 else "healthy"
 
 
+def _live_cooldown_seconds(cd: dict | None, now: datetime) -> float | None:
+    """Seconds until the next probe if this engine is *currently* benched.
+
+    None when not cooling, when the snapshot is stale (scraper likely down), or
+    when the probe is already due (remaining <= 0) — in those cases the
+    log-derived health label stands on its own.
+    """
+    if not cd or cd["failures"] <= 0 or cd["next_probe_at"] is None:
+        return None
+    if (now - cd["updated_at"]).total_seconds() > _COOLDOWN_STALE_SECONDS:
+        return None
+    remaining = (cd["next_probe_at"] - now).total_seconds()
+    return remaining if remaining > 0 else None
+
+
+def _empty_engine_row(engine: str) -> dict:
+    """A zeroed aggregate row for an engine with no scrapes in the window.
+
+    Lets an engine that is benched (hence producing no logs) still appear on the
+    monitor — otherwise a long cooldown would make it vanish from the table.
+    """
+    return {
+        "engine": engine,
+        "scrapes": 0,
+        "successes": 0,
+        "entries_parsed": 0,
+        "zero_parse": 0,
+        "failures": 0,
+        "blocked": 0,
+        "avg_latency_ms": None,
+        "p50_latency_ms": None,
+        "p95_latency_ms": None,
+        "last_scrape_at": None,
+        "last_success": None,
+        "last_http_status": None,
+        "last_error_message": None,
+        "http_status_breakdown": {},
+    }
+
+
 class OverallMetrics(BaseModel):
     scrapes: int = Field(..., description="Page-attempts in the window")
     successes: int = Field(..., description="Successful page-attempts")
@@ -87,7 +136,10 @@ class EngineMetrics(BaseModel):
     engine: str = Field(..., description="Search engine name")
     health: str = Field(
         ...,
-        description="healthy | degraded | blocked | parsing | idle (see classify_engine)",
+        description=(
+            "healthy | degraded | blocked | parsing | idle (see classify_engine), "
+            "or cooldown when the scraper has the engine benched"
+        ),
     )
     scrapes: int
     successes: int
@@ -107,6 +159,15 @@ class EngineMetrics(BaseModel):
     http_status_breakdown: dict[str, int] = Field(
         ..., description="Counts per monitored HTTP status (200/429/403/503)"
     )
+    cooldown_seconds_remaining: float | None = Field(
+        None,
+        description="Seconds until the scraper next probes a benched engine "
+        "(null when not cooling down)",
+    )
+    cooldown_failures: int = Field(
+        0,
+        description="Consecutive block signals driving the current cooldown (0 if none)",
+    )
 
 
 class CycleMetrics(BaseModel):
@@ -173,10 +234,23 @@ def _build_overall(row: dict) -> OverallMetrics:
     )
 
 
-def _build_engine(row: dict) -> EngineMetrics:
+def _build_engine(row: dict, cooldown: dict | None = None) -> EngineMetrics:
+    """Shape one engine row. ``cooldown`` is its live ``{failures, remaining}``
+    when currently benched, which overrides the log-derived health label."""
+    health = classify_engine(row)
+    cooldown_remaining = None
+    cooldown_failures = 0
+    if cooldown is not None:
+        cooldown_remaining = cooldown["remaining"]
+        cooldown_failures = cooldown["failures"]
+        # The engine is benched right now; that supersedes a stale log-based
+        # label (often "idle" — no logs are produced while skipped).
+        health = "cooldown"
     return EngineMetrics(
         engine=row["engine"],
-        health=classify_engine(row),
+        health=health,
+        cooldown_seconds_remaining=cooldown_remaining,
+        cooldown_failures=cooldown_failures,
         scrapes=row["scrapes"],
         successes=row["successes"],
         success_rate=_rate(row["successes"], row["scrapes"]),
@@ -209,9 +283,24 @@ async def get_metrics(
     metrics = await run_in_threadpool(db.get_scrape_metrics, window)
     cycles = await run_in_threadpool(db.get_recent_cycles, _RECENT_CYCLES)
     failures = await run_in_threadpool(db.get_recent_scrape_failures, _RECENT_FAILURES)
+    cooldowns = await run_in_threadpool(db.get_engine_cooldowns)
 
     overall = _build_overall(metrics["overall"])
-    engines = [_build_engine(row) for row in metrics["engines"]]
+
+    # Resolve which engines are *currently* benched (fresh snapshot, probe still
+    # in the future), then overlay that onto the per-engine rows.
+    now = _naive_utc_now()
+    live = {
+        engine: {"remaining": remaining, "failures": cd["failures"]}
+        for engine, cd in cooldowns.items()
+        if (remaining := _live_cooldown_seconds(cd, now)) is not None
+    }
+    rows = {row["engine"]: row for row in metrics["engines"]}
+    # A long cooldown means no scrapes in the window, so a benched engine may be
+    # absent from the aggregates entirely — synthesize a row so it still shows.
+    for engine in live:
+        rows.setdefault(engine, _empty_engine_row(engine))
+    engines = [_build_engine(rows[engine], live.get(engine)) for engine in sorted(rows)]
 
     return MetricsResponse(
         active_topics=len(topics),
diff --git a/common/block_signals.py b/common/block_signals.py
new file mode 100644
index 0000000..596cbde
--- /dev/null
+++ b/common/block_signals.py
@@ -0,0 +1,42 @@
+"""Shared block-signal detection used by both the scraper and the API.
+
+Some engines block at the network layer rather than with an HTTP status: the
+server (or an upstream proxy) tears down the TCP connection, so the browser
+nav fails with a ``net::ERR_CONNECTION_*`` error and there is *no* HTTP
+response to read a 429/403/503 from. Yahoo does exactly this — after a burst it
+serves a persistent empty HTTP 500 (Server: ATS, Connection: close) and a real
+browser nav then fails ``net::ERR_CONNECTION_CLOSED`` (see
+docs/BLOCK_SIGNAL_FINDINGS.md). Treating these as a block (not a transient
+glitch) lets the scraper bench the engine and the monitor label it correctly.
+
+Deliberately narrow: only connection-level *refusals/teardowns* count. A plain
+navigation timeout (``TimeoutError``) is a transient failure we neither punish
+(cooldown) nor flag (health) — it stays "other"/non-block, matching the
+existing contract in scraper/cooldown.py.
+"""
+
+# Substrings matched case-insensitively against a scrape's error message
+# (``"{ExceptionType}: {message}"``). These are Chromium/Playwright net errors
+# that mean the connection was refused or torn down before a usable response —
+# a server actively closing the door, i.e. a block.
+NETWORK_BLOCK_PATTERNS = (
+    "err_connection_closed",
+    "err_connection_reset",
+    "err_connection_refused",
+    "err_connection_aborted",
+    "err_connection_failed",
+    "err_socket_not_connected",
+    "err_empty_response",
+)
+
+
+def is_network_block(error_message: str | None) -> bool:
+    """True if ``error_message`` looks like a connection-level block/teardown.
+
+    Returns False for ``None``/empty and for transient errors such as
+    navigation timeouts.
+    """
+    if not error_message:
+        return False
+    haystack = error_message.lower()
+    return any(pattern in haystack for pattern in NETWORK_BLOCK_PATTERNS)
diff --git a/common/database.py b/common/database.py
index 145345a..4a68735 100644
--- a/common/database.py
+++ b/common/database.py
@@ -176,6 +176,16 @@ def ensure_schema() -> None:
                 "CREATE INDEX IF NOT EXISTS idx_scraper_cycles_started_at "
                 "ON scraper_cycles(started_at DESC)"
             )
+            cursor.execute(
+                """
+                CREATE TABLE IF NOT EXISTS engine_cooldowns (
+                    engine VARCHAR(32) PRIMARY KEY,
+                    failures INTEGER NOT NULL DEFAULT 0,
+                    next_probe_at TIMESTAMP,
+                    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+                )
+                """
+            )
         _schema_ready = True
 
 
@@ -665,6 +675,61 @@ def get_recent_cycles(limit: int = 24) -> list[dict]:
         return [dict(row) for row in cursor.fetchall()]
 
 
+# ── Engine cooldown snapshot (engine_cooldowns) ──────────────────────────────
+
+
+# Idempotent: each row is a full overwrite keyed by engine, so a retry after a
+# committed write just reapplies the same absolute state — safe.
+@retry_on_transient_error()
+def upsert_engine_cooldowns(
+    states: "list[tuple[str, int, float]]",
+) -> None:
+    """Persist the scraper's per-engine cooldown snapshot for the monitor.
+
+    ``states`` is ``(engine, failures, remaining_seconds)`` per tracked engine.
+    ``next_probe_at`` is stored as absolute UTC (``NOW() + remaining``) using the
+    *DB* clock so the API can derive a fresh countdown without trusting the
+    scraper's wall-clock; a non-cooling engine (failures == 0) stores NULL.
+    """
+    if not states:
+        return
+    with _Connection() as conn:
+        cursor = conn.cursor()
+        cursor.executemany(
+            "INSERT INTO engine_cooldowns (engine, failures, next_probe_at, updated_at) "
+            "VALUES (%s, %s, "
+            "        CASE WHEN %s > 0 THEN NOW() + make_interval(secs => %s) END, "
+            "        NOW()) "
+            "ON CONFLICT (engine) DO UPDATE SET "
+            "  failures = EXCLUDED.failures, "
+            "  next_probe_at = EXCLUDED.next_probe_at, "
+            "  updated_at = EXCLUDED.updated_at",
+            [(e, f, f, max(0.0, r)) for (e, f, r) in states],
+        )
+
+
+@retry_on_transient_error()
+def get_engine_cooldowns() -> dict[str, dict]:
+    """Per-engine cooldown state keyed by engine name (for the monitor).
+
+    Returns ``{engine: {failures, next_probe_at, updated_at}}``. The endpoint
+    decides what is still "live" (freshness + remaining); this is a plain read.
+    """
+    with _Connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT engine, failures, next_probe_at, updated_at FROM engine_cooldowns"
+        )
+        return {
+            row["engine"]: {
+                "failures": row["failures"],
+                "next_probe_at": row["next_probe_at"],
+                "updated_at": row["updated_at"],
+            }
+            for row in cursor.fetchall()
+        }
+
+
 # ── Scrape metrics aggregation (scraper_logs) ────────────────────────────────
 
 # Shared aggregate columns for the overall + per-engine rollups. duration_ms is
@@ -687,7 +752,8 @@ def get_recent_cycles(limit: int = 24) -> list[dict]:
     percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_latency_ms,
     MAX(scraped_at) AS last_scrape_at,
     (ARRAY_AGG(success ORDER BY scraped_at DESC))[1] AS last_success,
-    (ARRAY_AGG(http_status_code ORDER BY scraped_at DESC))[1] AS last_http_status
+    (ARRAY_AGG(http_status_code ORDER BY scraped_at DESC))[1] AS last_http_status,
+    (ARRAY_AGG(error_message ORDER BY scraped_at DESC))[1] AS last_error_message
 """
 
 _METRICS_WHERE = "scraped_at >= NOW() - make_interval(secs => %s)"
@@ -717,6 +783,7 @@ def _ms(value) -> int | None:
         "last_scrape_at": row["last_scrape_at"],
         "last_success": row["last_success"],
         "last_http_status": row["last_http_status"],
+        "last_error_message": row["last_error_message"],
         "http_status_breakdown": breakdown,
     }
 
diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md
index 0123c77..a08c359 100644
--- a/docs/API_REFERENCE.md
+++ b/docs/API_REFERENCE.md
@@ -249,8 +249,10 @@ Each `engines[*]` entry carries `scrapes`, `successes`, `success_rate`,
 `entries_parsed`, `zero_parse` (successful scrapes that parsed 0 items — a
 selector-rot signal), `failures`, `blocked` (failures with HTTP 429/403/503),
 `avg_latency_ms` / `p50_latency_ms` / `p95_latency_ms`, `last_scrape_at`,
-`last_success`, `last_http_status`, `http_status_breakdown`, and a heuristic
-`health` label: `healthy | degraded | blocked | parsing | idle` (see
+`last_success`, `last_http_status`, `http_status_breakdown`,
+`cooldown_seconds_remaining` (seconds until the scraper next probes a benched
+engine, else null) / `cooldown_failures`, and a heuristic `health` label:
+`healthy | degraded | blocked | parsing | cooldown | idle` (see
 [docs/OBSERVABILITY.md](OBSERVABILITY.md)). Each `recent_cycles[*]` carries
 `started_at`, `finished_at`, `duration_seconds`, `topics_count`,
 `entries_parsed`, `new_events`, `success`, `error`.
diff --git a/docs/BLOCK_SIGNAL_CHARACTERIZATION.md b/docs/BLOCK_SIGNAL_CHARACTERIZATION.md
index 7684281..b607a7f 100644
--- a/docs/BLOCK_SIGNAL_CHARACTERIZATION.md
+++ b/docs/BLOCK_SIGNAL_CHARACTERIZATION.md
@@ -68,8 +68,11 @@ data (2026-06-18 concurrency run):
   with `Connection: close`, persisting as an IP cooldown. A real browser nav
   then fails with `net::ERR_CONNECTION_CLOSED` — caught by the runner's outer
   exception handler as a failed scrape — *before* any body exists for
-  `detect_block` to inspect. So `detect_block` stays `None`; the honest-failure
-  path is the navigation-error + parse-0 nets, not a body signal. (Optional
+  `detect_block` to inspect. So `detect_block` (a *body* signal) stays `None`.
+  The connection-level teardown is instead recognized as a block by
+  `common/block_signals.is_network_block`, which feeds two places: the scraper's
+  adaptive cooldown benches Yahoo instead of re-hitting it every cycle, and the
+  monitor labels it `blocked` even though there is no HTTP status. (Optional
   hardening: add `500` to `monitored_codes` so a non-browser/200-path 500 is
   flagged too — see findings.)
 
diff --git a/docs/BLOCK_SIGNAL_FINDINGS.md b/docs/BLOCK_SIGNAL_FINDINGS.md
index 756f7f2..848ee10 100644
--- a/docs/BLOCK_SIGNAL_FINDINGS.md
+++ b/docs/BLOCK_SIGNAL_FINDINGS.md
@@ -195,15 +195,18 @@ Flooded Yahoo News at ~100–116 req/s. Yahoo served real results for exactly
 - **HTTP 500, empty body (`Content-Length: 0`)**, `Server: ATS` (Yahoo's Apache
   Traffic Server edge), `Cache-Control: private`, `Connection: close`. No
   redirect (`final_url` unchanged), no captcha copy — the body is literally
-  empty, so there is nothing for `detect_block` to match.
+  empty, so there is nothing for the body-level `detect_block` to match. (The
+  connection-level error *is* now treated as a block — see below.)
 - **Persistent / IP-scoped:** still 500 on low-rate sequential probes ~15 s
   after the flood stopped — a cooldown block of the IP, not momentary overload.
   (Contrast 2026-06-17: ~900 sequential requests never tripped it.)
 - **What the real (Playwright) scraper sees:** `page.goto` raises
   `net::ERR_CONNECTION_CLOSED` (the empty 500 + `Connection: close` looks like a
   dropped connection to Chromium). The runner's outer `except` catches it →
-  `success=False` with that error message. So the production scraper **already
-  fails honestly** on a Yahoo block; it just isn't labelled a "block".
+  `success=False` with that error message. `common/block_signals.is_network_block`
+  recognizes that error as a connection-level block, so the engine is now both
+  **benched by the adaptive cooldown** (rather than re-hit every cycle) and
+  **labelled `blocked` on the monitor** despite having no HTTP status.
 - Raw fixture saved: `scraper_dumps/yahoo_500_sample.html` (0 bytes — the empty
   body is itself the signal) and `yahoo_httpx_block_*` dumps.
 - `detect_block` action: **keep `None`** — there is no body to inspect (goto
@@ -275,9 +278,11 @@ Notes:
    resolved why their `detect_block` stubs should stay `None`: **Bing never
    hard-blocks** (silent per-IP throttle, ~50k requests all HTTP 200) and
    **Yahoo blocks with an empty HTTP 500** (no parseable challenge page; a real
-   browser sees `net::ERR_CONNECTION_CLOSED`, already caught as a nav error).
-   Neither exposes a body signal to key on. Optional: add `500` to
-   `monitored_codes` to flag a Yahoo 500 on the non-browser/200 path.
+   browser sees `net::ERR_CONNECTION_CLOSED`, caught as a nav error and now
+   recognized as a network-level block — `is_network_block` — so Yahoo is
+   benched and labelled `blocked`). Neither exposes a *body* signal to key on.
+   Optional: add `500` to `monitored_codes` to flag a Yahoo 500 on the
+   non-browser/200 path.
 4. The parse-0 / `_capture_diagnostic` self-documenting logging (plan step 5)
    worked: it captured Brave's captcha body and Google's `/sorry/` body
    automatically, with raw HTML dumped to `scraper_dumps/` for fixtures.
diff --git a/docs/OBSERVABILITY.md b/docs/OBSERVABILITY.md
index e01e6e2..c1fb97d 100644
--- a/docs/OBSERVABILITY.md
+++ b/docs/OBSERVABILITY.md
@@ -38,15 +38,24 @@ alongside it, so the label isn't the whole story.
 | Label      | Meaning                                                                           |
 | ---------- | --------------------------------------------------------------------------------- |
 | `idle`     | No scrapes for this engine in the window.                                         |
-| `blocked`  | The most recent scrape was a throttle/block (HTTP 429 / 403 / 503).               |
+| `blocked`  | The most recent scrape was a throttle/block: HTTP 429 / 403 / 503, or a connection-level teardown with no HTTP status (e.g. `ERR_CONNECTION_CLOSED`). |
+| `cooldown` | The scraper currently has the engine benched; shows a countdown to the next probe. |
 | `parsing`  | Sustained selector rot: ≥3 scrapes, ≥1 success, and every success parsed 0 items. |
 | `degraded` | Success rate below 0.75 (includes a total failure: 0%).                           |
 | `healthy`  | Otherwise.                                                                        |
 
 `blocked` keys off the **latest** scrape so an engine that recovered shows
 healthy, and one currently throttled shows blocked — even if its long-run
-success rate is high. `parsing` requires a sustained run (≥3 scrapes) so a
-single quiet hour for one topic doesn't trip it.
+success rate is high. A network-level block (no HTTP status) is recognized via
+`common/block_signals.is_network_block`. `parsing` requires a sustained run (≥3
+scrapes) so a single quiet hour for one topic doesn't trip it.
+
+`cooldown` is sourced differently from the others: it comes from the scraper's
+in-process `EngineCooldownTracker`, which the scraper snapshots to the
+`engine_cooldowns` table each cycle (the API can't see the live tracker across
+the process boundary). It overrides the log-derived label while an engine is
+benched — including engines that have produced no logs in the window and would
+otherwise drop off the table. A stale snapshot (scraper down) is ignored.
 
 ## What gets captured, and what `duration_ms` means
 
diff --git a/postgres/init.sql b/postgres/init.sql
index 02db424..593384b 100644
--- a/postgres/init.sql
+++ b/postgres/init.sql
@@ -91,6 +91,20 @@ CREATE TABLE IF NOT EXISTS scraper_cycles (
 );
 CREATE INDEX IF NOT EXISTS idx_scraper_cycles_started_at ON scraper_cycles(started_at DESC);
 
+-- Current adaptive cooldown state per engine, one row per engine. The scraper
+-- owns this table (it holds the live EngineCooldownTracker in-process, see
+-- scraper/cooldown.py) and overwrites a snapshot each cycle; the API only reads
+-- it to surface a "cooling down" indicator on the monitor page. next_probe_at is
+-- absolute UTC wall-clock (the tracker's monotonic clock can't cross processes),
+-- so the API derives the remaining seconds freshly. failures = 0 / NULL probe
+-- means the engine is not currently benched.
+CREATE TABLE IF NOT EXISTS engine_cooldowns (
+    engine VARCHAR(32) PRIMARY KEY,
+    failures INTEGER NOT NULL DEFAULT 0,
+    next_probe_at TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
 -- NOTIFY on each new feed event (topic match), not on news-content insert, so
 -- a topic referencing an already-stored article still streams to that topic.
 -- Format: "topic:topic_news_id".
diff --git a/scraper/cooldown.py b/scraper/cooldown.py
index 90a23af..6b56601 100644
--- a/scraper/cooldown.py
+++ b/scraper/cooldown.py
@@ -24,6 +24,7 @@
 from dataclasses import dataclass, field
 from typing import Callable, Literal
 
+from common.block_signals import is_network_block
 from common.model import ScraperLog
 
 logger = logging.getLogger(__name__)
@@ -36,9 +37,26 @@
 Decision = Literal["run", "probe", "skip"]
 
 
+@dataclass(frozen=True)
+class CooldownSnapshot:
+    """One engine's cooldown state, serialized for cross-process consumption.
+
+    ``remaining_seconds`` is the wall-clock-equivalent of the tracker's internal
+    monotonic ``next_probe`` at snapshot time, so a reader in another process
+    (the API) can turn it into an absolute timestamp.
+    """
+
+    engine: str
+    failures: int
+    remaining_seconds: float
+
+
 def _is_block_message(log: ScraperLog) -> bool:
-    # detect_block / redirected_off_results record "<engine> blocked: <reason>".
-    return "block" in (log.error_message or "").lower()
+    # detect_block / redirected_off_results record "<engine> blocked: <reason>";
+    # a connection-level teardown (e.g. Yahoo's ERR_CONNECTION_CLOSED) carries no
+    # HTTP status but is equally a block, so fold it in here too.
+    message = log.error_message or ""
+    return "block" in message.lower() or is_network_block(message)
 
 
 def classify_logs(logs: list[ScraperLog]) -> Outcome:
@@ -123,3 +141,15 @@ def record(self, engine: str, logs: list[ScraperLog]) -> None:
             # the same window so a flaky probe doesn't become a per-topic retry
             # storm across the rest of the cycle.
             state.next_probe = self._clock() + self._window_for(state.failures)
+
+    def snapshot(self) -> list[CooldownSnapshot]:
+        """Current state of every tracked engine, for persistence (see
+        db.upsert_engine_cooldowns). Engines never seen are simply absent."""
+        return [
+            CooldownSnapshot(
+                engine=engine,
+                failures=state.failures,
+                remaining_seconds=self.remaining(engine),
+            )
+            for engine, state in self._state.items()
+        ]
diff --git a/scraper/main.py b/scraper/main.py
index ee6e50f..6cc5e72 100644
--- a/scraper/main.py
+++ b/scraper/main.py
@@ -317,6 +317,20 @@ def main():
                     db.insert_scraper_logs(all_logs)
                     cycle_success = True
 
+                    # Publish the in-process cooldown state so the API/monitor
+                    # can show which engines are benched. Best-effort: a write
+                    # failure here must not fail the (otherwise successful) cycle.
+                    if cooldown is not None:
+                        try:
+                            db.upsert_engine_cooldowns(
+                                [
+                                    (s.engine, s.failures, s.remaining_seconds)
+                                    for s in cooldown.snapshot()
+                                ]
+                            )
+                        except Exception:
+                            logger.exception("Failed to publish engine cooldown state")
+
                 except Exception as e:
                     cycle_error = f"{type(e).__name__}: {e}"
                     logger.error(f"Error in scraping loop: {e}")
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 0666bad..c559dda 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -55,7 +55,7 @@ def _clean(db):
     """Reset all rows (and serial counters) before each test for isolation."""
     with db._Connection() as conn:
         conn.cursor().execute(
-            "TRUNCATE scraper_cycles, topic_news, news, scraper_logs, topics "
-            "RESTART IDENTITY CASCADE"
+            "TRUNCATE scraper_cycles, topic_news, news, scraper_logs, "
+            "engine_cooldowns, topics RESTART IDENTITY CASCADE"
         )
     yield
diff --git a/tests/integration/test_api_integration.py b/tests/integration/test_api_integration.py
index 107d46a..00c559e 100644
--- a/tests/integration/test_api_integration.py
+++ b/tests/integration/test_api_integration.py
@@ -212,6 +212,41 @@ def test_metrics_with_scraper_activity(client, db):
     assert body["recent_failures"][0]["engine"] == "bing"
 
 
+def test_metrics_blocked_on_connection_closed(client, db):
+    # A connection-level teardown (no HTTP status) is classified as blocked.
+    db.add_topic("alpha")
+    with db._Connection() as conn:
+        conn.cursor().execute(
+            "INSERT INTO scraper_logs "
+            "(topic, scraped_at, success, http_status_code, error_message, engine) "
+            "VALUES ('alpha', NOW(), FALSE, NULL, "
+            "'Error: Page.goto: net::ERR_CONNECTION_CLOSED at https://y/s', 'yahoo')"
+        )
+
+    body = client.get("/api/v1/metrics?window=3600").json()
+    yahoo = {e["engine"]: e for e in body["engines"]}["yahoo"]
+    assert yahoo["health"] == "blocked"
+
+
+def test_metrics_surfaces_cooldown(client, db):
+    # An engine that is benched but produced no logs in the window still appears,
+    # labeled "cooldown" with a countdown to the next probe.
+    db.add_topic("alpha")
+    db.upsert_engine_cooldowns([("brave", 2, 286.0), ("google", 0, 0.0)])
+
+    body = client.get("/api/v1/metrics?window=3600").json()
+    by_engine = {e["engine"]: e for e in body["engines"]}
+
+    # brave: synthesized row (no scrapes), shown as cooling down.
+    assert by_engine["brave"]["health"] == "cooldown"
+    assert by_engine["brave"]["cooldown_failures"] == 2
+    assert 0 < by_engine["brave"]["cooldown_seconds_remaining"] <= 286
+    assert by_engine["brave"]["scrapes"] == 0
+
+    # google: failures == 0 → not cooling, so it isn't surfaced from an empty row.
+    assert "google" not in by_engine
+
+
 def test_response_has_timing_header(client):
     r = client.get("/api/v1/topics")
     assert "x-process-time-ms" in {k.lower() for k in r.headers}
diff --git a/tests/test_engine_cooldown.py b/tests/test_engine_cooldown.py
index 8a6093e..9d8d38c 100644
--- a/tests/test_engine_cooldown.py
+++ b/tests/test_engine_cooldown.py
@@ -59,6 +59,19 @@ def test_classify_other_for_non_block_failure():
     assert classify_logs(logs) == "other"
 
 
+def test_classify_block_from_connection_closed():
+    # Yahoo-style network teardown (no HTTP status) counts as a block, so the
+    # engine gets benched instead of retried every cycle.
+    logs = [
+        _log(
+            "yahoo",
+            ok=False,
+            error="Error: Page.goto: net::ERR_CONNECTION_CLOSED at https://y/search",
+        )
+    ]
+    assert classify_logs(logs) == "block"
+
+
 # --- tracker decide/record ------------------------------------------------
 
 
@@ -121,6 +134,30 @@ def test_transient_failure_during_probe_rearms_without_deepening():
     assert round(tracker.remaining("g")) == 300
 
 
+# --- snapshot -------------------------------------------------------------
+
+
+def test_snapshot_reports_failures_and_remaining():
+    clock = _Clock()
+    tracker = _tracker(clock, base=300)
+    tracker.record("yahoo", [_log("yahoo", ok=False, status=503)])
+    tracker.record("google", [_log("google", ok=True, n=2)])
+
+    snap = {s.engine: s for s in tracker.snapshot()}
+    assert set(snap) == {"yahoo", "google"}
+
+    assert snap["yahoo"].failures == 1
+    assert 0 < snap["yahoo"].remaining_seconds <= 300
+
+    # A healthy engine is still reported, but not cooling.
+    assert snap["google"].failures == 0
+    assert snap["google"].remaining_seconds == 0.0
+
+
+def test_snapshot_empty_before_any_activity():
+    assert _tracker(_Clock()).snapshot() == []
+
+
 # --- scrape_topic integration ---------------------------------------------
 
 
diff --git a/tests/test_metrics.py b/tests/test_metrics.py
index af95d34..34ee16a 100644
--- a/tests/test_metrics.py
+++ b/tests/test_metrics.py
@@ -4,7 +4,14 @@
 pure Python derivation that turns an aggregate row into a triage label.
 """
 
-from api.v1.metrics import _rate, classify_engine
+from datetime import datetime, timedelta
+
+from api.v1.metrics import (
+    _COOLDOWN_STALE_SECONDS,
+    _live_cooldown_seconds,
+    _rate,
+    classify_engine,
+)
 
 
 def _row(
@@ -16,6 +23,7 @@ def _row(
     blocked=0,
     last_success=True,
     last_http_status=200,
+    last_error_message=None,
 ):
     return {
         "scrapes": scrapes,
@@ -25,6 +33,7 @@ def _row(
         "blocked": blocked,
         "last_success": last_success,
         "last_http_status": last_http_status,
+        "last_error_message": last_error_message,
     }
 
 
@@ -69,6 +78,34 @@ def test_not_blocked_when_latest_ok_despite_block_history():
     assert classify_engine(r) == "healthy"
 
 
+def test_blocked_on_connection_closed_without_http_status():
+    # Yahoo-style network teardown: no HTTP status, but the error message is a
+    # connection-level block → "blocked", not "degraded".
+    r = _row(
+        scrapes=4,
+        successes=0,
+        last_success=False,
+        last_http_status=None,
+        last_error_message=(
+            "Error: Page.goto: net::ERR_CONNECTION_CLOSED at "
+            "https://news.search.yahoo.com/search?p=spacex&b=1"
+        ),
+    )
+    assert classify_engine(r) == "blocked"
+
+
+def test_timeout_is_not_a_network_block():
+    # A navigation timeout is transient, not a block: stays degraded.
+    r = _row(
+        scrapes=4,
+        successes=0,
+        last_success=False,
+        last_http_status=None,
+        last_error_message="TimeoutError: Timeout 30000ms exceeded.",
+    )
+    assert classify_engine(r) == "degraded"
+
+
 def test_parsing_when_all_successes_parse_zero():
     r = _row(
         scrapes=6, successes=6, zero_parse=6, last_success=True, last_http_status=200
@@ -115,3 +152,43 @@ def test_rate_helper():
     assert _rate(8, 10) == 0.8
     assert _rate(0, 0) is None
     assert _rate(3, 4) == 0.75
+
+
+# ── _live_cooldown_seconds ───────────────────────────────────────────────────
+
+_NOW = datetime(2026, 6, 18, 12, 0, 0)
+
+
+def _cd(*, failures=1, probe_in=300, updated_ago=5):
+    """A cooldown row: probe `probe_in`s from _NOW, snapshot `updated_ago`s old."""
+    return {
+        "failures": failures,
+        "next_probe_at": _NOW + timedelta(seconds=probe_in),
+        "updated_at": _NOW - timedelta(seconds=updated_ago),
+    }
+
+
+def test_cooldown_live_when_benched_and_fresh():
+    assert _live_cooldown_seconds(_cd(probe_in=286), _NOW) == 286
+
+
+def test_cooldown_none_when_not_cooling():
+    assert _live_cooldown_seconds(_cd(failures=0), _NOW) is None
+    assert _live_cooldown_seconds(None, _NOW) is None
+    assert (
+        _live_cooldown_seconds(
+            {"failures": 1, "next_probe_at": None, "updated_at": _NOW}, _NOW
+        )
+        is None
+    )
+
+
+def test_cooldown_none_when_probe_already_due():
+    # next_probe_at in the past: the engine probes next cycle, not benched now.
+    assert _live_cooldown_seconds(_cd(probe_in=-10), _NOW) is None
+
+
+def test_cooldown_none_when_snapshot_stale():
+    # Scraper hasn't refreshed the snapshot recently → ignore (likely down).
+    stale = _cd(probe_in=300, updated_ago=_COOLDOWN_STALE_SECONDS + 60)
+    assert _live_cooldown_seconds(stale, _NOW) is None