Merge pull request #158 from magnus919/fix/raleigh-rpd-live-queries

fix(raleigh): restore live RPD queries

raleigh/EVIDENCE-LEDGER.md

@@ -190,3 +190,42 @@ The following public service boundaries were exercised successfully on 2026-07-2
 - Upstream node IDs, headings, table headers, publication labels, or origins may change. Detectable changes fail visibly; revise the adapter and fixtures only after re-verifying the official source contract.
 - No model-backed eval run, CI run, commit, push, pull request, deployment, release, or merge is claimed.
 - Roll back the aggregate adapter and CLI wiring if the official site removes JSON:API access or publication links cannot be validated without broadening the trust boundary.
+
+## Issue 157 Addendum: Restore Live RPD Queries
+
+### Intent and authority
+
+- Restore the date-filtered NIBRS and CrimeMapper query paths that failed against their live ArcGIS layers.
+- Add deterministic regression coverage and bounded scheduled canary probes for both affected sources.
+- Modify the local Raleigh skill only. No publish, deploy, merge, authentication, or write authority was used.
+
+### Root cause and decision
+
+- Live metadata resolved NIBRS item `24c0b37fa9bb4e16ba8bcaa7e806c615` and CrimeMapper item `a1f2d9204a184404b5a4c7e0fdceb6d0` to queryable layer `0` with the expected date fields and query capabilities.
+- Both layers rejected bare epoch-millisecond date comparisons such as `reported_date >= 1700000000000` with `Invalid query parameters`, while the equivalent UTC `TIMESTAMP 'YYYY-MM-DD HH:MM:SS'` predicate succeeded. The working previous-day control did not add a date predicate.
+- Police date filters now use the same bounded ArcGIS timestamp-literal conversion already proven by the fire adapter. Raw user filter values are not logged.
+- The scheduled canary now verifies the required date field and a non-empty one-record, date-filtered response for NIBRS and CrimeMapper so this live contract is checked independently of mocked fixtures. Exhausted transport failures also fail the workflow after bounded retries instead of producing a false-green run.
+
+### Files changed
+
+- Updated `scripts/raleighlib/police.py`, `scripts/canary.py`, and `tests/test_raleigh.py`.
+- Added deterministic assertions for NIBRS and SRS timestamp predicates, out-of-range epochs, both bounded canary calls, police schema drift, and exhausted canary transport failures.
+
+### Verification
+
+- `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest raleigh.tests.test_raleigh.PoliceTests`: **35 tests passed**.
+- `PYTHONDONTWRITEBYTECODE=1 python3 -m unittest raleigh/tests/test_raleigh.py`: **373 tests passed**.
+- `python3 -m ruff check raleigh/scripts/raleighlib/police.py`: passed.
+- `python3 scripts/validate-evals.py raleigh`: **12 eval manifests validated**.
+- `ruby scripts/validate-skills.rb`: **111 canonical skills validated**.
+- `git diff --check`: passed.
+- Live `police incidents --since 30d --category burglary --limit 3`: returned three NIBRS burglary records.
+- Live `police recent --limit 3`: returned three CrimeMapper records.
+- Live `police history --reporting-system nibrs --since 30d --limit 3`: returned three NIBRS records.
+- Direct `probe_police()`: passed one-record probes for both NIBRS and CrimeMapper.
+
+### Remaining boundaries
+
+- The full scheduled GitHub Actions canary was not run locally; its new police probe function was exercised directly against both live services.
+- Future ArcGIS schema or SQL-dialect changes remain outside this verification window and should surface through the scheduled canary.
+- No commit, push, pull request, CI run, deployment, release, or merge is claimed.

raleigh/scripts/canary.py

@@ -36,6 +36,7 @@ from raleighlib import transit
 from raleighlib import development
 from raleighlib import civic
 from raleighlib import meetings
+from raleighlib import police
 
 MAX_RETRIES = 2
 RETRY_DELAY_SECONDS = 5
@@ -368,6 +369,73 @@ def probe_imagery_catalog() -> list[dict[str, Any]]:
     return results
 
 
+def probe_police() -> list[dict[str, Any]]:
+    """Exercise bounded date-filtered queries against fixed RPD sources."""
+    results: list[dict[str, Any]] = []
+    for source_key in ("nibrs", "crimemapper-90d"):
+        layer_url, err = _probe_with_retry(police.resolve_layer_url, source_key)
+        if err:
+            results.append({"source": "police", "target": source_key, "status": "fail", **err})
+            continue
+        fields, err = _probe_with_retry(arcgis.layer_fields, layer_url)
+        if err:
+            results.append({"source": "police", "target": source_key, "status": "fail", **err})
+            continue
+        field_names = {
+            field.get("name") for field in fields if isinstance(field, dict)
+        }
+        if "reported_date" not in field_names:
+            results.append({
+                "source": "police",
+                "target": source_key,
+                "status": "fail",
+                "failure_class": "schema_drift",
+                "error": "missing required field: reported_date",
+                "attempt": 1,
+            })
+            continue
+        collection, err = _probe_with_retry(
+            police.query_incidents,
+            source_key,
+            since_ms=police.NIBRS_EPOCH_MS,
+            limit=1,
+        )
+        if err:
+            results.append({"source": "police", "target": source_key, "status": "fail", **err})
+            continue
+        if (
+            not isinstance(collection, dict)
+            or collection.get("type") != "FeatureCollection"
+            or not isinstance(collection.get("features"), list)
+        ):
+            results.append({
+                "source": "police",
+                "target": source_key,
+                "status": "fail",
+                "failure_class": "schema_drift",
+                "error": "expected GeoJSON FeatureCollection",
+                "attempt": 1,
+            })
+            continue
+        if not collection["features"]:
+            results.append({
+                "source": "police",
+                "target": source_key,
+                "status": "fail",
+                "failure_class": "schema_drift",
+                "error": "date-filtered query returned no records",
+                "attempt": 1,
+            })
+            continue
+        results.append({
+            "source": "police",
+            "target": source_key,
+            "status": "pass",
+            "count": len(collection["features"]),
+        })
+    return results
+
+
 ALL_PROBES = [
     ("hub-catalog", probe_hub_catalog),
     ("geocode", probe_geocode),
@@ -377,6 +445,7 @@ ALL_PROBES = [
     ("civic-rss", probe_civic_rss),
     ("meetings", probe_meetings),
     ("imagery", probe_imagery_catalog),
+    ("police", probe_police),
 ]
 
 
@@ -415,7 +484,7 @@ def run_canary() -> dict[str, Any]:
         else:
             durable_failures += 1
 
-    passed = durable_failures == 0
+    passed = durable_failures == 0 and transient_failures == 0
     report = {
         "canary": "raleigh-live-endpoint",
         "started_at": started,

raleigh/scripts/raleighlib/police.py

@@ -98,6 +98,15 @@ def _discover_fields(layer_url: str) -> set[str]:
     return {f.get("name", "") for f in fields if isinstance(f, dict)}
 
 
+def _ms_to_timestamp_literal(ms: int) -> str:
+    """Format Unix milliseconds as an ArcGIS TIMESTAMP literal in UTC."""
+    try:
+        dt = datetime.fromtimestamp(ms / 1000, timezone.utc)
+    except (OverflowError, OSError, ValueError) as exc:
+        raise PoliceError(f"date range out of bounds: {ms}") from exc
+    return "TIMESTAMP '" + dt.strftime("%Y-%m-%d %H:%M:%S") + "'"
+
+
 def build_where_clause(
     source_key: str,
     available_fields: set[str],
@@ -119,7 +128,7 @@ def build_where_clause(
     if since_ms is not None:
         date_field = field_map["date"]
         if date_field in available_fields:
-            clauses.append(f"{date_field} >= {since_ms}")
+            clauses.append(f"{date_field} >= {_ms_to_timestamp_literal(since_ms)}")
         else:
             print(
                 f"Warning: date field '{date_field}' not found in {source_key}; skipping date filter",

raleigh/tests/test_raleigh.py

@@ -46,6 +46,7 @@ import raleighlib.fire as fire
 import raleighlib.fire_protection as fire_protection
 import raleighlib.rfd_reports as rfd_reports
 import raleighlib.public_safety_stats as public_safety_stats
+import canary as canary_lib
 from raleighlib import cli as cli_lib
 
 CLI_SCRIPT = _SCRIPT_DIR / "raleigh"
@@ -2461,7 +2462,11 @@ class PoliceTests(unittest.TestCase):
 
     def test_date_filter_builds_where_clause(self):
         where = police.build_where_clause("nibrs", self.NIBRS_FIELDS, since_ms=1700000000000)
-        self.assertIn("reported_date >= 1700000000000", where)
+        self.assertIn("reported_date >= TIMESTAMP '2023-11-14 22:13:20'", where)
+
+    def test_date_filter_rejects_out_of_range_epoch(self):
+        with self.assertRaisesRegex(police.PoliceError, "date range out of bounds"):
+            police.build_where_clause("nibrs", self.NIBRS_FIELDS, since_ms=10**30)
 
     def test_category_filter_quotes_value(self):
         where = police.build_where_clause("nibrs", self.NIBRS_FIELDS, category="BURGLARY")
@@ -2482,7 +2487,7 @@ class PoliceTests(unittest.TestCase):
     def test_srs_uses_different_field_names(self):
         where = police.build_where_clause("srs", self.SRS_FIELDS, category="LARCENY", since_ms=1700000000000)
         self.assertIn("UPPER(LCR_DESC) LIKE '%LARCENY%'", where)
-        self.assertIn("INC_DATETIME >= 1700000000000", where)
+        self.assertIn("INC_DATETIME >= TIMESTAMP '2023-11-14 22:13:20'", where)
 
     def test_missing_field_skips_filter_with_warning(self):
         sparse_fields = {"OBJECTID"}
@@ -2629,6 +2634,50 @@ class PoliceTests(unittest.TestCase):
         self.assertEqual(code, 0)
         self.assertIn("predates NIBRS", err.getvalue())
 
+    def test_canary_probes_date_filtered_nibrs_and_crimemapper_queries(self):
+        collection = {"type": "FeatureCollection", "features": [{"properties": {}}]}
+        with patch("canary.police.resolve_layer_url", return_value="https://example.test/0"), patch(
+            "canary.arcgis.layer_fields", return_value=[{"name": "reported_date"}]
+        ), patch("canary.police.query_incidents", return_value=collection) as query:
+            results = canary_lib.probe_police()
+        self.assertEqual([result["target"] for result in results], ["nibrs", "crimemapper-90d"])
+        self.assertTrue(all(result["status"] == "pass" for result in results))
+        self.assertEqual(query.call_count, 2)
+        for call in query.call_args_list:
+            self.assertEqual(call.kwargs["since_ms"], police.NIBRS_EPOCH_MS)
+            self.assertEqual(call.kwargs["limit"], 1)
+
+    def test_canary_rejects_police_source_without_date_field(self):
+        with patch("canary.police.resolve_layer_url", return_value="https://example.test/0"), patch(
+            "canary.arcgis.layer_fields", return_value=[{"name": "OBJECTID"}]
+        ), patch("canary.police.query_incidents") as query:
+            results = canary_lib.probe_police()
+        self.assertTrue(all(result["status"] == "fail" for result in results))
+        self.assertTrue(all(result["failure_class"] == "schema_drift" for result in results))
+        query.assert_not_called()
+
+    def test_canary_rejects_empty_police_query(self):
+        collection = {"type": "FeatureCollection", "features": []}
+        with patch("canary.police.resolve_layer_url", return_value="https://example.test/0"), patch(
+            "canary.arcgis.layer_fields", return_value=[{"name": "reported_date"}]
+        ), patch("canary.police.query_incidents", return_value=collection):
+            results = canary_lib.probe_police()
+        self.assertTrue(all(result["status"] == "fail" for result in results))
+        self.assertTrue(all(result["failure_class"] == "schema_drift" for result in results))
+
+    def test_canary_fails_after_exhausted_transport_error(self):
+        failure = [{
+            "source": "test",
+            "target": "endpoint",
+            "status": "fail",
+            "failure_class": "transport_outage",
+            "error": "timed out",
+        }]
+        with patch.object(canary_lib, "ALL_PROBES", [("test", lambda: failure)]):
+            report = canary_lib.run_canary()
+        self.assertFalse(report["passed"])
+        self.assertEqual(report["summary"]["transient_failures"], 1)
+
 
 class FireTests(unittest.TestCase):
     """Tests for the RFD incident command group and 2026 schema normalization."""