Harden the feed download against paging, empty and pinless responses

CLAUDE.md

@@ -47,7 +47,8 @@ with the evidence that closed them.
 | Duration outliers are categorical, not statistical. The 14-day cap is a backstop, not the outlier strategy. | data-quality.md — "Duration outliers are categorical" |
 | "We are investigating" reference pairing works but rescues almost nothing — not worth building. | data-quality.md — "'We are investigating' notices" (corrected 2026-07-20) |
 | `closed_at` is a floor: short-lived cases are never observed open. Twice-daily builds are the settled cadence. | data-quality.md — "`closed_at` is a floor" (re-measured 2026-07-31) |
-| A case the feed drops while `Open` is stamped `vanished_at` (schema v4) and is closed with no signal on the site, never `closed_at`. The stamp is safe only behind the feed-count guard (`FEED_COUNT_TOLERANCE`), which refuses a short download before anything touches the DB. | data-quality.md - "Cases that vanish from the feed" (2026-09-05) |
+| A case the feed drops while `Open` is stamped `vanished_at` (schema v4) and is closed with no signal on the site, never `closed_at`; the stamp touches closed rows too. It is safe only behind the feed-count guard (`FEED_COUNT_TOLERANCE`), which refuses a short download before anything touches the DB, and behind the empty-download refusal. Paging is by `OBJECTID`, refused unless each page is strictly ascending. | data-quality.md - "Cases that vanish from the feed" (2026-09-05, amended 2026-09-24) |
+| A feature with no pin (no `geometry`, or `"NaN"`) keeps the pin the DB last stored, or is set aside with a `::warning::` until the feed pins it. Nullable coordinates were rejected: not an additive migration. | data-quality.md - "A feature with no pin" (2026-09-24) |
 | **A case is open only while nothing its own text has ended.** `is_open(row, now)` reads `status`, `vanished_at` and a passed *observed* end, decided once in `resolve_case` and carried on `Case.is_open` for every surface that says open. The close date follows the same reading: the notice's own completion, else `closed_at` (`closed_on`, 2026-09-24). The feed closes a case a median 72h after the notice reports completion; 216 of 562 `Open` cases were past one, 0 of 7,667 completions were ever followed up. Scheduled ends do not close a case for display. | statuspage-methodology.md - "The notice's own completion closes it" (2026-09-05) |
 | gemma-4-12b-qat over qwen3.5-9b for end-time extraction; prompt version is at v3. | model-and-runtime-benchmarks.md, end-time-eval.md |
 | Geography is CSO Census settlements, not the feed's `location` string (3,866 distinct values, fragments badly, carries no population). | statuspage-methodology.md — "The county drill-down" (2026-07-25) |

notes/data-quality.md

@@ -294,6 +294,29 @@ and refuses a download more than 1% short (`FEED_COUNT_TOLERANCE`). The 1% is fo
 changing under the paging; a real purge like the one around 2026-04-20 would still be stamped,
 which is right: that is what happened.
 
+*2026-09-24:* the tolerance passes an empty feed, 0 downloaded of 0 reported, and that build
+would stamp vanished every row not yet vanished, closed ones included (4,306 on the 2026-09-23
+release, 498 of them open). `run` now also refuses an empty download
+while the DB holds rows not yet vanished; the guard counts what the stamp touches, not the
+open ones only. No real purge has emptied the feed; the 2026-08-10 one left 3,044 cases, so a
+partial purge is still stamped as before. The same review moved `download_cases` from
+`resultOffset` to `OBJECTID > <last seen>` paging: by offset, one case deleted during the
+download pushed a live case out of the next page and stamped it vanished, and a server
+`maxRecordCount` below the 1,000 asked for dropped the difference at every page, both inside
+the 1%. Key paging is only sound on pages returned in `OBJECTID` order, so a page that is not
+ascending fails the build rather than skip or repeat rows.
+
+### A feature with no pin (2026-09-24)
+
+ArcGIS omits `geometry` for a null shape, or writes an empty point as `"NaN"`, and one such
+feature crashed every build: the coordinates are `NOT NULL` in `cases` and key the geocode
+cache. Making them nullable was rejected, because that is not an additive migration.
+`restore_pins` instead gives a case the pin the DB last stored for it, and sets aside a case
+the DB has never seen pinned, with a `::warning::` on the Actions run naming its id: such a
+case is missing from the site, health notices included, until the feed pins it. It is not in
+the DB, so it cannot be stamped vanished. None of the 13,588 cases on the 2026-09-23 release
+lacked a pin.
+
 The first v4 build will stamp all 9,053 (verified on a copy of the release: the stamp is
 idempotent across builds and clears when a case returns), and `create_db` prints the count
 stamped on every build from now on, so the next purge is in the build log the day it happens.

src/uisce/pipeline.py

@@ -1,12 +1,15 @@
 import argparse
 import json
+import math
 import os
 import re
 import sqlite3
 import time
 from collections import Counter
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from itertools import pairwise
 from pathlib import Path
 
 import requests
@@ -35,6 +38,7 @@
 LOCATIONIQ_REVERSE_URL = "https://us1.locationiq.com/v1/reverse"
 LOCATIONIQ_GEOCODE_SLEEP = 1
 COORD_PRECISION = 4  # ~10 meter
+COORD_COLUMNS = ("full_lat", "full_lon", "rounded_lat", "rounded_lon")
 
 USABLE_CASE_THRESHOLD_FIELDS = ["TITLE", "DESCRIPTION"]
 
@@ -181,24 +185,55 @@ def feed_count(session):
     return data["count"]
 
 
-def check_download_complete(features, expected):
+@contextmanager
+def _cases_table(db_path):
+    """(read-only connection, cases columns), or (None, set()) while db_path holds
+    no cases table (geocode_all can create the file before create_db has run)."""
+    if not db_path.exists():
+        yield None, set()
+        return
+    conn = sqlite3.connect(f"{Path(db_path).resolve().as_uri()}?mode=ro", uri=True)
+    try:
+        columns = {row[1] for row in conn.execute("PRAGMA table_info(cases)")}
+        yield (conn if columns else None), columns
+    finally:
+        conn.close()
+
+
+def unvanished_cases(db_path=DB_PATH):
+    """Rows load_cases would stamp vanished if the download held none of them."""
+    with _cases_table(db_path) as (conn, columns):
+        if conn is None:
+            return 0
+        live = " WHERE vanished_at IS NULL" if "vanished_at" in columns else ""
+        return conn.execute(f"SELECT COUNT(*) FROM cases{live}").fetchone()[0]
+
+
+def check_download_complete(features, expected, unvanished=0):
     if len(features) < expected * (1 - FEED_COUNT_TOLERANCE):
         raise RuntimeError(
             f"downloaded {len(features)} cases but the feed reports {expected}; "
             "refusing to build from a truncated download"
         )
+    # the tolerance passes 0 of 0, and an empty download stamps every stored case vanished
+    if unvanished and not features:
+        raise RuntimeError(
+            f"the download is empty (the feed reports {expected}) but the DB holds "
+            f"{unvanished} cases not yet vanished; refusing to stamp them all"
+        )
 
 
 def download_cases(session):
     all_features = []
-    offset = 0
+    # By key, not offset: a row deleted mid-download shifts every later offset,
+    # and a server maxRecordCount below the page size drops rows at each page.
+    last_id = -1
 
     while True:
         params = {
-            "where": "1=1",
+            "where": f"OBJECTID > {last_id}",
             "outFields": "*",
             "orderByFields": "OBJECTID",
-            "resultOffset": offset,
             "resultRecordCount": ARCGIS_PAGE_SIZE,
             "f": "json",
         }
@@ -208,19 +243,26 @@ def download_cases(session):
         data = resp.json()
 
         if "error" in data:
-            raise RuntimeError(f"ArcGIS error at offset {offset}: {data['error']}")
+            raise RuntimeError(f"ArcGIS error after OBJECTID {last_id}: {data['error']}")
 
         features = data.get("features", [])
         if not features:
             break
 
+        ids = [(f.get("attributes") or {}).get("OBJECTID") for f in features]
+        # key paging is only sound on ids strictly above the last page's, in order
+        if None in ids or not all(a < b for a, b in pairwise([last_id, *ids])):
+            raise RuntimeError(
+                f"ArcGIS page after OBJECTID {last_id} is not strictly ascending: "
+                f"{ids[:5]}{'...' if len(ids) > 5 else ''}"
+            )
         all_features.extend(features)
         print(f"Fetched {len(all_features)}")
 
         if not data.get("exceededTransferLimit", False):
             break
 
-        offset += ARCGIS_PAGE_SIZE
+        last_id = ids[-1]
         time.sleep(ARCGIS_PAGE_SLEEP)
 
     print(f"Done: {len(all_features)} records")
@@ -264,12 +306,18 @@ def map_cases(cases_to_map):
         mapped_case["start_date"] = _epoch_ms_to_iso(mapped_case["start_date"])
         mapped_case["end_date"] = _epoch_ms_to_iso(mapped_case["end_date"])
 
-        lon, lat = transformer.transform(case["geometry"]["x"], case["geometry"]["y"])
-        mapped_case["full_lat"] = lat
-        mapped_case["full_lon"] = lon
-
-        mapped_case["rounded_lat"] = round(lat, COORD_PRECISION)
-        mapped_case["rounded_lon"] = round(lon, COORD_PRECISION)
+        # ArcGIS omits `geometry` for a null shape, or writes an empty point as
+        # "NaN"; restore_pins settles both
+        geometry = case.get("geometry") or {}
+        x, y = _coordinate(geometry.get("x")), _coordinate(geometry.get("y"))
+        if x is None or y is None:
+            mapped_case.update(dict.fromkeys(COORD_COLUMNS))
+        else:
+            lon, lat = transformer.transform(x, y)
+            mapped_case["full_lat"] = lat
+            mapped_case["full_lon"] = lon
+            mapped_case["rounded_lat"] = round(lat, COORD_PRECISION)
+            mapped_case["rounded_lon"] = round(lon, COORD_PRECISION)
 
         if mapped_case["county"] == "Dnegal":
             mapped_case["county"] = "Donegal"
@@ -284,6 +332,40 @@ def map_cases(cases_to_map):
     return all_cases, skipped
 
 
+def _coordinate(value):
+    try:
+        value = float(value)
+    except (TypeError, ValueError):
+        return None
+    return value if math.isfinite(value) else None
+
+
+def restore_pins(cases, db_path=DB_PATH):
+    """(cases, unplaced ids): a pinless case gets its stored pin or is set aside.
+    See notes/data-quality.md, "A feature with no pin" (2026-09-24)."""
+    missing = [c["id"] for c in cases if c["full_lat"] is None]
+    if not missing:
+        return cases, []
+    stored = {}
+    with _cases_table(db_path) as (conn, _):
+        if conn is not None:
+            marks = ", ".join("?" * len(missing))
+            rows = conn.execute(
+                f"SELECT id, {', '.join(COORD_COLUMNS)} FROM cases WHERE id IN ({marks})",
+                missing,
+            )
+            stored = {row[0]: dict(zip(COORD_COLUMNS, row[1:])) for row in rows}
+    kept, unplaced = [], []
+    for case in cases:
+        if case["full_lat"] is not None:
+            kept.append(case)
+        elif case["id"] in stored:
+            kept.append(case | stored[case["id"]])
+        else:
+            unplaced.append(case["id"])
+    return kept, unplaced
+
+
 def _epoch_ms_to_iso(ms):
     if ms is None:
         return None
@@ -872,7 +954,8 @@ def backfill_reduced_pressure(conn):
     every other backfill.
     """
     rows = conn.execute(
-        "SELECT id, description FROM cases WHERE description IS NOT NULL AND NOT reduced_pressure"
+        "SELECT id, description FROM cases "
+        "WHERE description IS NOT NULL AND COALESCE(reduced_pressure, 0) = 0"
     ).fetchall()
     updates = [
         (case_id,) for case_id, description in rows if _PRESSURE_ONLY.search(description)
@@ -940,13 +1023,18 @@ def run(skip_geocode=False):
     session = make_session()
     expected = feed_count(session)
     features = download_cases(session)
-    check_download_complete(features, expected)
+    check_download_complete(features, expected, unvanished_cases())
     CASES_RAW_PATH.parent.mkdir(parents=True, exist_ok=True)
     CASES_RAW_PATH.write_text(json.dumps(features, indent=2))
 
     mapped_cases, skipped = map_cases(read_arcgis_cases())
     if skipped:
         print(f"Skipped {len(skipped)} cases with no usable data: {skipped}")
+    mapped_cases, unplaced = restore_pins(mapped_cases)
+    if unplaced:
+        # surfaced on the Actions run: these are live notices missing from the site
+        print(f"::warning::{len(unplaced)} cases the feed has never pinned are left out "
+              f"until it does: {unplaced}")
     CASES_MAPPED_PATH.parent.mkdir(parents=True, exist_ok=True)
     CASES_MAPPED_PATH.write_text(json.dumps(mapped_cases, indent=2))