Sync from GitHub · 2026-10-21 11:53 UTC

intel/README.md

@@ -0,0 +1,88 @@
+# Intel store (`backend/intel/`)
+
+This is Phase 0 and Phase 1 of the predictive intelligence system. The design is in
+[docs/blueprint/intelligence-system.md](../../docs/blueprint/intelligence-system.md), and the decisions
+are in ADR-0017 to ADR-0021. It runs locally against a SQLite file and is not wired into the deployed API.
+The only touchpoint is the news → events hook, which does nothing unless `INTEL_DB_PATH` is set.
+
+## Set up
+
+All commands run from `backend/`.
+
+1. Install the dependencies:
+
+   ```bash
+   py -m pip install -r requirements.txt -r requirements-intel.txt
+   ```
+
+2. Add the store's path to `backend/.env`:
+
+   ```
+   INTEL_DB_PATH=intel/data/intel.db
+   ```
+
+3. Download the raw files into `intel/data/raw/`. The folder is gitignored.
+   - **Boundaries** come from geoBoundaries (CC BY 4.0). Use the simplified NGA ADM1 and ADM2 files,
+     named `geoBoundaries-NGA-ADM1_simplified.geojson` and `geoBoundaries-NGA-ADM2_simplified.geojson`.
+     The download links are on <https://www.geoboundaries.org/countryDownloads.html> (NGA).
+   - **Overture places** (CDLA Permissive 2.0) are extracted with duckdb straight from Overture's
+     public S3 bucket. It takes about 30 s and produces 156k rows:
+
+     ```bash
+     py -m intel.ingest.overture_extract
+     ```
+
+## Build, in order
+
+```bash
+py -m intel.cli init                 # schema + source/stream registry
+py -m intel.cli locations            # gazetteer: 37 states, 774 LGAs, aliases
+py -m intel.cli calendar             # holidays and seasons, 2000 to today+10y
+py -m intel.cli places               # Overture → places, subsector mapping, locality aliases
+py -m intel.cli weather-forecast     # 16-day forecast per state
+py -m intel.cli weather-history      # resumable backfill from 2006, 4,500 weighted calls per run
+```
+
+Re-running any of these is safe, because every build is idempotent. The full weather history is about
+20,000 weighted Open-Meteo calls. The free tier allows 10,000 a day, so run `weather-history` once a day
+until `--estimate` reports 0. Each run resumes where the last one stopped.
+
+- **Licence warning:** Open-Meteo's free tier is **non-commercial only**. Commercial use needs a paid
+  plan or ECMWF open data (ADR-0020).
+- **One writer at a time:** SQLite takes a single writer, so don't run two commands against the store
+  at once.
+
+## Day to day
+
+```bash
+py -m intel.cli status               # Layer 1 report; exits 1 if anything isn't fresh
+py -m intel.cli resolve "Ajah"       # gazetteer lookup → ng.lagos.eti-osa
+py -m intel.cli run-daily            # forecast + history top-up + status, under one run id
+py -m intel.cli export               # CSVs + manifest for the Colab notebooks → intel/data/export/
+```
+
+`run-daily` is what a scheduler will call. There is no scheduler yet: the design names a GitHub
+Actions cron or Supabase `pg_cron`, because the Hugging Face Space sleeps when idle.
+
+## Layout
+
+| File | Job |
+|---|---|
+| `db.py` | Connection and schema init (`../sql/intel_schema.sql`) |
+| `registry.py` | Sources (cadence, latency, licence) and streams |
+| `handoff.py` | Layer names, handoff receipts, `ACTIVE_LAYERS` |
+| `freshness.py` | Layer 1: the two-clock source grading and overdue handoffs |
+| `geo.py` | Pure-Python centroid, point-in-polygon and polygon index |
+| `locations.py` | Gazetteer build, boundary matching, `resolve()` |
+| `subsector_map.py` | Overture categories → VenThrive's 49 subsectors |
+| `events.py` | News → permanent `events` hook (ADR-0015 addendum) |
+| `export.py` | CSV export for `ml/` |
+| `ingest/` | One module per source |
+
+## Tests
+
+```bash
+py -m pytest tests/test_intel.py -q
+```
+
+The tests are pure and offline: they use an in-memory database and fake HTTP fetches.

intel/cli.py

@@ -0,0 +1,108 @@
+"""
+Command line for the intel store (run from backend/ with INTEL_DB_PATH set):
+
+    py -m intel.cli init                     create tables, seed sources/streams
+    py -m intel.cli locations                build the state/LGA gazetteer
+    py -m intel.cli calendar                 holidays + seasons, 2000 → +10 years
+    py -m intel.cli places                   load the Overture extract (run overture_extract first)
+    py -m intel.cli weather-history [--budget N] [--start YYYY-MM-DD]   resumable backfill
+    py -m intel.cli weather-forecast         16-day forecast at state centres
+    py -m intel.cli status                   Layer 1 report (exit 1 if unhealthy)
+    py -m intel.cli export [--out DIR]       CSVs + manifest for the Colab notebooks
+    py -m intel.cli resolve "TEXT"           which location does this text mention?
+    py -m intel.cli run-daily                forecast + history top-up + status, one run id
+
+`run-daily` is what a scheduler will call once one exists (none yet — see the
+design doc's "Conflicts").
+"""
+import argparse
+import json
+import sys
+import uuid
+from datetime import date
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from intel import db  # noqa: E402  (env must be loaded first)
+
+EXPORT_DIR = Path(__file__).resolve().parent / "data" / "export"
+
+
+def _print(obj) -> None:
+    print(json.dumps(obj, indent=2, default=str))
+
+
+def main(argv: list[str] | None = None) -> int:
+    p = argparse.ArgumentParser(prog="intel")
+    sub = p.add_subparsers(dest="cmd", required=True)
+    sub.add_parser("init")
+    sub.add_parser("locations")
+    sub.add_parser("calendar")
+    sub.add_parser("places")
+    wh = sub.add_parser("weather-history")
+    wh.add_argument("--budget", type=float, default=None)
+    wh.add_argument("--start", type=date.fromisoformat, default=None)
+    wh.add_argument("--estimate", action="store_true", help="only print the weighted calls still needed")
+    sub.add_parser("weather-forecast")
+    sub.add_parser("status")
+    ex = sub.add_parser("export")
+    ex.add_argument("--out", type=Path, default=EXPORT_DIR)
+    rs = sub.add_parser("resolve")
+    rs.add_argument("text")
+    sub.add_parser("run-daily")
+    args = p.parse_args(argv)
+
+    con = db.connect()
+    run_id = f"{args.cmd}-{uuid.uuid4().hex[:8]}"
+    try:
+        if args.cmd == "init":
+            db.init(con)
+            _print({"initialised": db.db_path()})
+        elif args.cmd == "locations":
+            from intel.locations import build
+            _print(build(con))
+        elif args.cmd == "calendar":
+            from intel.ingest.calendar import ingest
+            _print(ingest(con, run_id=run_id))
+        elif args.cmd == "places":
+            from intel.ingest.places import ingest
+            _print(ingest(con, run_id=run_id))
+        elif args.cmd == "weather-history":
+            from intel.ingest import weather
+            start = args.start or weather.DEFAULT_HISTORY_START
+            if args.estimate:
+                _print({"weighted_calls_needed": round(weather.estimate_backfill_calls(con, start=start), 1)})
+            else:
+                _print(weather.backfill_history(con, start=start, budget=args.budget or weather.DEFAULT_RUN_BUDGET, run_id=run_id))
+        elif args.cmd == "weather-forecast":
+            from intel.ingest.weather import fetch_forecast
+            _print(fetch_forecast(con, run_id=run_id))
+        elif args.cmd == "status":
+            from intel.freshness import report
+            r = report(con)
+            _print(r)
+            return 0 if r["healthy"] else 1
+        elif args.cmd == "export":
+            from intel.export import export
+            _print(export(con, args.out)["rows"])
+        elif args.cmd == "resolve":
+            from intel.locations import resolve
+            _print({"text": args.text, "location_id": resolve(con, args.text)})
+        elif args.cmd == "run-daily":
+            from intel.freshness import report
+            from intel.ingest import weather
+            out = {"run_id": run_id, "forecast": weather.fetch_forecast(con, run_id=run_id),
+                   "history": weather.backfill_history(con, run_id=run_id)}
+            out["status"] = report(con)
+            _print(out)
+            return 0 if out["status"]["healthy"] else 1
+    finally:
+        con.close()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

intel/db.py

@@ -0,0 +1,49 @@
+"""
+Connection to the intel store. Phase 1 runs on a local SQLite file (ADR-0020);
+the schema in backend/sql/intel_schema.sql is portable so the same tables move
+to Postgres/Supabase later.
+
+INTEL_DB_PATH selects the file. When it's unset the store is "off": the CLI
+refuses to run and hooks from other systems (the news pipeline's event writer)
+silently do nothing — so the deployed backend, which has no intel store yet,
+behaves exactly as before.
+"""
+import os
+import sqlite3
+from datetime import datetime, timezone
+from pathlib import Path
+
+SCHEMA_PATH = Path(__file__).resolve().parent.parent / "sql" / "intel_schema.sql"
+
+
+def db_path() -> str | None:
+    return os.environ.get("INTEL_DB_PATH") or None
+
+
+def enabled() -> bool:
+    return db_path() is not None
+
+
+def connect(path: str | None = None) -> sqlite3.Connection:
+    path = path or db_path()
+    if not path:
+        raise RuntimeError("INTEL_DB_PATH is not set — the intel store is off (see backend/intel/README.md).")
+    if path != ":memory:":
+        Path(path).parent.mkdir(parents=True, exist_ok=True)
+    con = sqlite3.connect(path)
+    con.row_factory = sqlite3.Row
+    con.execute("pragma foreign_keys = on")
+    if path != ":memory:":
+        con.execute("pragma journal_mode = wal")
+    return con
+
+
+def init(con: sqlite3.Connection) -> None:
+    con.executescript(SCHEMA_PATH.read_text(encoding="utf8"))
+    from intel.registry import seed
+    seed(con)
+    con.commit()
+
+
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds")

intel/events.py

@@ -0,0 +1,70 @@
+"""
+News → permanent event history. The news pipeline (ADR-0015) decides what users
+*see* and lets items expire from the feed; this keeps every published item as a
+dated, located event because, for the correlation step, past events are the
+history — the reason the `threat` value was removed (no value in showing
+after-the-fact news) doesn't apply to *learning* from it.
+
+record_news_item() is called from the news pipeline and the admin publish route.
+It never raises and is a no-op when the intel store is off (INTEL_DB_PATH unset),
+so the deployed backend is unaffected.
+"""
+from datetime import date, datetime
+
+from intel import db
+from intel.locations import resolve
+
+SOURCE_ID = "news_pipeline"
+# news categories → event family; identical names except where noted.
+CATEGORY_TO_FAMILY = {
+    "policy": "policy", "weather": "weather", "infrastructure": "infrastructure",
+    "funding": "funding", "security": "security", "other": "other",
+}
+
+
+def _day(value) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, (date, datetime)):
+        return value.date().isoformat() if isinstance(value, datetime) else value.isoformat()
+    return str(value)[:10]
+
+
+def record_news_item(item: dict) -> bool:
+    """item: a news_items-shaped dict (url_hash, title, raw_text, category,
+    location, published_at, scraped_at, deadline). Returns True if an event was
+    written. Only published (status 'classified') items should be passed in."""
+    if not db.enabled():
+        return False
+    try:
+        con = db.connect()
+        try:
+            location_id = resolve(con, item.get("location")) or resolve(con, item.get("title"))
+            con.execute(
+                """insert into events (id, family, type, location_id, starts_on, ends_on, title, detail,
+                                       is_estimate, source_id, source_ref, ingested_at)
+                   values (?, ?, 'news', ?, ?, ?, ?, ?, 0, ?, ?, ?)
+                   on conflict (source_id, source_ref) do update set family = excluded.family,
+                     location_id = excluded.location_id, title = excluded.title, detail = excluded.detail""",
+                (
+                    f"{SOURCE_ID}:{item['url_hash']}",
+                    CATEGORY_TO_FAMILY.get(item.get("category") or "other", "other"),
+                    location_id,
+                    _day(item.get("published_at")) or _day(item.get("scraped_at")),
+                    _day(item.get("deadline")),
+                    item["title"],
+                    (item.get("raw_text") or "")[:2000] or None,
+                    SOURCE_ID,
+                    item["url_hash"],
+                    db.now_iso(),
+                ),
+            )
+            con.commit()
+            from intel.freshness import record_attempt
+            record_attempt(con, SOURCE_ID, ok=True, latest_data_period=_day(item.get("published_at")) or _day(item.get("scraped_at")))
+            return True
+        finally:
+            con.close()
+    except Exception as e:
+        print(f"[intel] could not record news event {item.get('url_hash')}: {e}")
+        return False

intel/export.py

@@ -0,0 +1,45 @@
+"""
+Exports the store as gzipped CSVs plus a manifest, for the Colab notebooks in
+ml/notebooks/. The manifest carries Layer 1's freshness report, so every
+notebook run states what data it was trained on and how old it was.
+"""
+import csv
+import gzip
+import json
+import sqlite3
+from pathlib import Path
+
+from intel.db import now_iso
+from intel.freshness import report
+
+TABLES = {
+    "locations": "select id, level, name, parent_id, lat, lon from locations order by id",
+    "streams": "select * from streams order by id",
+    "observations": """select stream_id, location_id, period_start, value from observations
+                       where is_forecast = 0 order by stream_id, location_id, period_start""",
+    "forecasts": """select stream_id, location_id, period_start, issued_at, value from observations
+                    where is_forecast = 1 order by issued_at, stream_id, location_id, period_start""",
+    "events": """select id, family, type, location_id, starts_on, ends_on, title, is_estimate, source_id
+                 from events order by starts_on""",
+    "places": """select id, name, category, subsector, confidence, operating_status, lat, lon, location_id, locality
+                 from places order by location_id""",
+}
+
+
+def export(con: sqlite3.Connection, out_dir: Path) -> dict:
+    out_dir.mkdir(parents=True, exist_ok=True)
+    counts = {}
+    for name, sql in TABLES.items():
+        cur = con.execute(sql)
+        cols = [d[0] for d in cur.description]
+        n = 0
+        with gzip.open(out_dir / f"{name}.csv.gz", "wt", newline="", encoding="utf8") as f:
+            w = csv.writer(f)
+            w.writerow(cols)
+            for row in cur:
+                w.writerow(row)
+                n += 1
+        counts[name] = n
+    manifest = {"exported_at": now_iso(), "rows": counts, "freshness": report(con)}
+    (out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf8")
+    return manifest

intel/freshness.py

@@ -0,0 +1,118 @@
+"""
+Layer 1 — stale awareness. Answers one question: is every source and every
+handoff where it should be right now? It reads; it never fixes anything.
+
+A source is judged against its own rhythm (registry.py), on two clocks:
+  - fetch clock: when did we last successfully fetch it?
+  - data clock:  how new is the newest period it gave us, allowing for the
+                 source's legitimate latency?
+A fetch that succeeds daily but keeps returning the same old period is stale on
+the data clock even though the fetch clock looks fine.
+"""
+import sqlite3
+from datetime import datetime, timedelta, timezone
+
+from intel.db import now_iso
+from intel.handoff import ACTIVE_LAYERS
+
+# A source is 'late' after 1.5× its cadence and 'stale' after 3×. Generous on
+# purpose: a monthly source a few days late is normal, three months late isn't.
+LATE_FACTOR = 1.5
+STALE_FACTOR = 3.0
+# A handoff to an active layer that sits unreceived this long is overdue.
+HANDOFF_OVERDUE = timedelta(hours=24)
+
+STATUS_ORDER = ["never", "failing", "stale", "late", "fresh"]
+
+
+def _parse(ts: str | None) -> datetime | None:
+    if not ts:
+        return None
+    dt = datetime.fromisoformat(ts)
+    return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
+
+
+def record_attempt(con: sqlite3.Connection, source_id: str, ok: bool,
+                   latest_data_period: str | None = None, error: str | None = None) -> None:
+    """Every ingester calls this once per run, success or failure."""
+    now = now_iso()
+    if ok:
+        con.execute(
+            """update sources set last_attempt_at = ?, last_success_at = ?, last_error = null,
+                 latest_data_period = coalesce(?, latest_data_period) where id = ?""",
+            (now, now, latest_data_period, source_id),
+        )
+    else:
+        con.execute("update sources set last_attempt_at = ?, last_error = ? where id = ?", (now, error, source_id))
+    con.commit()
+
+
+def _grade(age: timedelta, cadence: timedelta) -> str:
+    if age > cadence * STALE_FACTOR:
+        return "stale"
+    if age > cadence * LATE_FACTOR:
+        return "late"
+    return "fresh"
+
+
+def source_status(row: sqlite3.Row, now: datetime) -> dict:
+    cadence = timedelta(hours=row["expected_cadence_hours"])
+    latency = timedelta(hours=row["latency_hours"])
+    last_success = _parse(row["last_success_at"])
+    last_attempt = _parse(row["last_attempt_at"])
+    out = {
+        "source": row["id"], "family": row["family"],
+        "last_success_at": row["last_success_at"], "latest_data_period": row["latest_data_period"],
+        "last_error": row["last_error"],
+    }
+
+    if last_success is None:
+        out["status"] = "failing" if last_attempt else "never"
+        out["reason"] = f"never succeeded: {row['last_error']}" if last_attempt else "never fetched"
+        return out
+
+    fetch_grade = _grade(now - last_success, cadence)
+    data_grade = "fresh"
+    period = _parse(row["latest_data_period"])
+    if period is not None:
+        data_grade = _grade(max(now - period - latency, timedelta(0)), cadence)
+
+    # A success clears last_error, so an error present means the latest attempt
+    # failed — the source is failing now, even if its last good data is still
+    # within cadence. (Comparing timestamps instead misses a failure in the same
+    # second as the success.)
+    failing = bool(row["last_error"])
+    grade = min(fetch_grade, data_grade, key=STATUS_ORDER.index)
+    if failing:
+        out["status"], out["reason"] = "failing", f"last attempt failed: {row['last_error']}"
+    else:
+        out["status"] = grade
+        if grade == "fresh":
+            out["reason"] = "on schedule"
+        elif STATUS_ORDER.index(data_grade) < STATUS_ORDER.index(fetch_grade):
+            out["reason"] = f"fetched fine but newest data is {row['latest_data_period']}"
+        else:
+            out["reason"] = f"last successful fetch {row['last_success_at']}"
+    return out
+
+
+def report(con: sqlite3.Connection, now: datetime | None = None) -> dict:
+    now = now or datetime.now(timezone.utc)
+    sources = [source_status(r, now) for r in con.execute("select * from sources order by id")]
+
+    overdue, queued = [], []
+    for h in con.execute("select id, run_id, from_layer, to_layer, produced_at from handoffs where received_at is null"):
+        item = dict(h)
+        if h["to_layer"] not in ACTIVE_LAYERS:
+            queued.append(item)
+        elif now - _parse(h["produced_at"]) > HANDOFF_OVERDUE:
+            overdue.append(item)
+
+    healthy = all(s["status"] == "fresh" for s in sources) and not overdue
+    return {
+        "checked_at": now.isoformat(timespec="seconds"),
+        "healthy": healthy,
+        "sources": sources,
+        "handoffs_overdue": overdue,
+        "handoffs_queued_for_unbuilt_layers": len(queued),
+    }

intel/geo.py

@@ -0,0 +1,97 @@
+"""
+Pure-Python geometry for the intel store — GeoJSON Polygon/MultiPolygon only.
+Enough for Phase 1 (LGA centroids, which-LGA-is-this-point-in, distances) without
+pulling a GIS stack into the backend; PostGIS replaces this when the store moves
+to Postgres (ADR-0020).
+"""
+import math
+
+EARTH_RADIUS_KM = 6371.0088
+
+
+def _rings(geometry: dict) -> list[list[list[list[float]]]]:
+    """Normalises Polygon and MultiPolygon to a list of polygons, each a list of
+    rings (outer ring first, then holes), each ring a list of [lon, lat]."""
+    if geometry["type"] == "Polygon":
+        return [geometry["coordinates"]]
+    if geometry["type"] == "MultiPolygon":
+        return geometry["coordinates"]
+    raise ValueError(f"Unsupported geometry type: {geometry['type']}")
+
+
+def _ring_area_centroid(ring: list[list[float]]) -> tuple[float, float, float]:
+    """Signed area and area-weighted centroid of one ring (shoelace formula)."""
+    a = cx = cy = 0.0
+    for (x0, y0), (x1, y1) in zip(ring, ring[1:] + ring[:1]):
+        cross = x0 * y1 - x1 * y0
+        a += cross
+        cx += (x0 + x1) * cross
+        cy += (y0 + y1) * cross
+    a /= 2.0
+    if a == 0:
+        xs, ys = [p[0] for p in ring], [p[1] for p in ring]
+        return 0.0, sum(xs) / len(xs), sum(ys) / len(ys)
+    return a, cx / (6.0 * a), cy / (6.0 * a)
+
+
+def centroid(geometry: dict) -> tuple[float, float]:
+    """Area-weighted centroid as (lon, lat). Holes subtract their area. Planar
+    maths on degrees is fine at LGA scale this close to the equator."""
+    total = sx = sy = 0.0
+    for polygon in _rings(geometry):
+        for i, ring in enumerate(polygon):
+            a, cx, cy = _ring_area_centroid([p[:2] for p in ring])
+            a = abs(a) if i == 0 else -abs(a)
+            total += a
+            sx += cx * a
+            sy += cy * a
+    if total == 0:
+        raise ValueError("Degenerate geometry")
+    return sx / total, sy / total
+
+
+def bbox(geometry: dict) -> tuple[float, float, float, float]:
+    xs, ys = [], []
+    for polygon in _rings(geometry):
+        for p in polygon[0]:
+            xs.append(p[0])
+            ys.append(p[1])
+    return min(xs), min(ys), max(xs), max(ys)
+
+
+def _in_ring(lon: float, lat: float, ring: list[list[float]]) -> bool:
+    inside = False
+    for (x0, y0), (x1, y1) in zip(ring, ring[1:] + ring[:1]):
+        if (y0 > lat) != (y1 > lat) and lon < (x1 - x0) * (lat - y0) / (y1 - y0) + x0:
+            inside = not inside
+    return inside
+
+
+def contains(geometry: dict, lon: float, lat: float) -> bool:
+    for polygon in _rings(geometry):
+        outer, holes = polygon[0], polygon[1:]
+        if _in_ring(lon, lat, [p[:2] for p in outer]) and not any(_in_ring(lon, lat, [p[:2] for p in h]) for h in holes):
+            return True
+    return False
+
+
+def haversine_km(lon1: float, lat1: float, lon2: float, lat2: float) -> float:
+    p1, p2 = math.radians(lat1), math.radians(lat2)
+    dp, dl = p2 - p1, math.radians(lon2 - lon1)
+    h = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
+    return 2 * EARTH_RADIUS_KM * math.asin(math.sqrt(h))
+
+
+class PolygonIndex:
+    """Point-in-polygon lookup over many features, with a bbox prefilter so a
+    lookup only runs the ray-cast against the handful of polygons whose box
+    contains the point."""
+
+    def __init__(self, features: list[tuple[str, dict]]):
+        self._items = [(key, geom, bbox(geom)) for key, geom in features]
+
+    def lookup(self, lon: float, lat: float) -> str | None:
+        for key, geom, (x0, y0, x1, y1) in self._items:
+            if x0 <= lon <= x1 and y0 <= lat <= y1 and contains(geom, lon, lat):
+                return key
+        return None

intel/handoff.py

@@ -0,0 +1,48 @@
+"""
+Handoff receipts between layers. A layer's job ends with produce(); the next
+layer's job starts with receive(). Neither layer knows anything else about the
+other — and Layer 1 can see any report that was produced but never received.
+"""
+import json
+import sqlite3
+import uuid
+
+from intel.db import now_iso
+
+STALE_AWARENESS = "stale_awareness"
+DATA_STREAM = "data_stream"
+INVESTIGATION = "investigation"
+INTERFACE = "interface"
+SEARCH = "search"
+
+# Layers that exist and run. A report addressed to a layer that isn't built yet
+# is queued, not overdue — Layer 1 lists it separately instead of raising it.
+ACTIVE_LAYERS = {STALE_AWARENESS, DATA_STREAM}
+
+
+def produce(con: sqlite3.Connection, run_id: str, from_layer: str, to_layer: str, report: dict) -> str:
+    hid = uuid.uuid4().hex
+    con.execute(
+        "insert into handoffs (id, run_id, from_layer, to_layer, report, produced_at) values (?, ?, ?, ?, ?, ?)",
+        (hid, run_id, from_layer, to_layer, json.dumps(report, default=str), now_iso()),
+    )
+    con.commit()
+    return hid
+
+
+def receive(con: sqlite3.Connection, handoff_id: str, by: str) -> bool:
+    """Acknowledges a report. Returns False if it was already received (or
+    doesn't exist), so a receiver can tell a duplicate from a first delivery."""
+    cur = con.execute(
+        "update handoffs set received_at = ?, received_by = ? where id = ? and received_at is null",
+        (now_iso(), by, handoff_id),
+    )
+    con.commit()
+    return cur.rowcount == 1
+
+
+def pending(con: sqlite3.Connection, to_layer: str) -> list[dict]:
+    rows = con.execute(
+        "select * from handoffs where to_layer = ? and received_at is null order by produced_at", (to_layer,)
+    ).fetchall()
+    return [{**dict(r), "report": json.loads(r["report"])} for r in rows]

intel/ingest/calendar.py

@@ -0,0 +1,116 @@
+"""
+Nigeria's national calendar as events: federal public holidays, the Islamic
+holidays and Ramadan (computed, so marked is_estimate), and the December festive
+trading season. Generated rather than scraped, so history back to 2000 costs
+nothing — which matters, because the correlation step learns from history.
+
+Known limits, kept honest by is_estimate and the design doc:
+  - Islamic dates use the Umm al-Qura calendar; Nigeria's dates follow the
+    Sultan of Sokoto's moon-sighting announcement and can differ by a day.
+  - The Federal Government sometimes declares extra or substitute days (e.g. a
+    Monday when a holiday falls on Sunday, or a one-off holiday). Those aren't
+    predictable; the news pipeline is where declarations will be caught.
+  - School terms vary by state and aren't included yet.
+"""
+import sqlite3
+from datetime import date, timedelta
+
+from hijridate import Gregorian, Hijri
+
+from intel.db import now_iso
+from intel.freshness import record_attempt
+from intel.handoff import DATA_STREAM, INTERFACE, produce
+
+SOURCE_ID = "calendar"
+COUNTRY = "ng"
+FIRST_YEAR = 2000
+YEARS_AHEAD = 10
+
+
+def easter(year: int) -> date:
+    """Western Easter Sunday (anonymous Gregorian algorithm)."""
+    a, b, c = year % 19, year // 100, year % 100
+    d, e = b // 4, b % 4
+    f = (b + 8) // 25
+    g = (b - f + 1) // 3
+    h = (19 * a + b - d - g + 15) % 30
+    i, k = c // 4, c % 4
+    l = (32 + 2 * e + 2 * i - h - k) % 7
+    m = (a + 11 * h + 22 * l) // 451
+    month = (h + l - 7 * m + 114) // 31
+    day = (h + l - 7 * m + 114) % 31 + 1
+    return date(year, month, day)
+
+
+def _hijri_to_greg(hy: int, hm: int, hd: int) -> date:
+    return Hijri(hy, hm, hd).to_gregorian()
+
+
+def _hijri_years_overlapping(year: int) -> range:
+    """The Hijri years whose dates can fall in this Gregorian year (a Hijri year
+    is ~11 days shorter, so it's always the one in progress on 1 January and the
+    next)."""
+    hy = Gregorian(year, 1, 1).to_hijri().year
+    return range(hy, hy + 2)
+
+
+def events_for_year(year: int) -> list[dict]:
+    """[{key, type, title, starts_on, ends_on, is_estimate}] for one Gregorian year."""
+    out = []
+
+    def add(key, type_, title, start, end=None, estimate=False):
+        if start.year == year or (end and end.year == year):
+            out.append({"key": key, "type": type_, "title": title, "starts_on": start, "ends_on": end or start,
+                        "is_estimate": estimate})
+
+    add("new_year", "public_holiday", "New Year's Day", date(year, 1, 1))
+    gf = easter(year) - timedelta(days=2)
+    add("good_friday", "public_holiday", "Good Friday", gf)
+    add("easter_monday", "public_holiday", "Easter Monday", gf + timedelta(days=3))
+    add("workers_day", "public_holiday", "Workers' Day", date(year, 5, 1))
+    # Democracy Day moved from 29 May to 12 June from 2019.
+    add("democracy_day", "public_holiday", "Democracy Day", date(year, 6, 12) if year >= 2019 else date(year, 5, 29))
+    add("independence_day", "public_holiday", "Independence Day", date(year, 10, 1))
+    add("christmas", "public_holiday", "Christmas Day", date(year, 12, 25))
+    add("boxing_day", "public_holiday", "Boxing Day", date(year, 12, 26))
+    # Festive trading season: VenThrive's own window for the December spending peak.
+    add("festive_season", "season", "Festive trading season", date(year, 12, 15), date(year + 1, 1, 5), estimate=True)
+
+    for hy in _hijri_years_overlapping(year):
+        ramadan_start = _hijri_to_greg(hy, 9, 1)
+        eid_fitr = _hijri_to_greg(hy, 10, 1)
+        add("ramadan", "season", "Ramadan", ramadan_start, eid_fitr - timedelta(days=1), estimate=True)
+        add("eid_al_fitr", "religious_holiday", "Eid al-Fitr", eid_fitr, eid_fitr + timedelta(days=1), estimate=True)
+        eid_adha = _hijri_to_greg(hy, 12, 10)
+        add("eid_al_adha", "religious_holiday", "Eid al-Adha", eid_adha, eid_adha + timedelta(days=1), estimate=True)
+        add("mawlid", "religious_holiday", "Eid-el-Maulud (Mawlid)", _hijri_to_greg(hy, 3, 12), estimate=True)
+    return out
+
+
+def ingest(con: sqlite3.Connection, first_year: int = FIRST_YEAR, last_year: int | None = None,
+           run_id: str = "manual") -> dict:
+    last_year = last_year or date.today().year + YEARS_AHEAD
+    now = now_iso()
+    n = 0
+    try:
+        for year in range(first_year, last_year + 1):
+            for e in events_for_year(year):
+                ref = f"{e['key']}:{e['starts_on'].isoformat()}"
+                con.execute(
+                    """insert into events (id, family, type, location_id, starts_on, ends_on, title, detail,
+                                           is_estimate, source_id, source_ref, ingested_at)
+                       values (?, 'calendar', ?, ?, ?, ?, ?, null, ?, ?, ?, ?)
+                       on conflict (source_id, source_ref) do update set title = excluded.title,
+                         ends_on = excluded.ends_on, is_estimate = excluded.is_estimate""",
+                    (f"{SOURCE_ID}:{ref}", e["type"], COUNTRY, e["starts_on"].isoformat(), e["ends_on"].isoformat(),
+                     e["title"], int(e["is_estimate"]), SOURCE_ID, ref, now),
+                )
+                n += 1
+        con.commit()
+    except Exception as e:
+        record_attempt(con, SOURCE_ID, ok=False, error=str(e))
+        raise
+    record_attempt(con, SOURCE_ID, ok=True, latest_data_period=f"{last_year}-12-31")
+    summary = {"years": f"{first_year}-{last_year}", "events": n}
+    produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "calendar", **summary})
+    return summary

intel/ingest/overture_extract.py

@@ -0,0 +1,63 @@
+"""
+One-time (per Overture release) extract of Nigeria's places from Overture's public
+S3 parquet into a local parquet file. ingest/places.py then loads that file into
+the intel store — the slow network read happens once, not on every ingest.
+
+    py -m intel.ingest.overture_extract [--release 2026-09-23.1]
+
+Filtered by Nigeria's bounding box only; the exact country boundary is applied at
+ingest (point-in-LGA-polygon), which also drops the bits of Benin/Cameroon/Niger the
+box catches.
+"""
+import argparse
+import time
+from pathlib import Path
+
+import duckdb
+
+DEFAULT_RELEASE = "2026-09-23.1"
+# Nigeria's extent with a small margin (lon 2.67–14.68, lat 4.27–13.89).
+BBOX = {"xmin": 2.6, "xmax": 14.7, "ymin": 4.2, "ymax": 13.95}
+RAW_DIR = Path(__file__).resolve().parent.parent / "data" / "raw"
+
+
+def extract_path(release: str) -> Path:
+    return RAW_DIR / f"overture_places_ng_{release}.parquet"
+
+
+def extract(release: str = DEFAULT_RELEASE) -> Path:
+    out = extract_path(release)
+    RAW_DIR.mkdir(parents=True, exist_ok=True)
+    con = duckdb.connect()
+    con.sql("INSTALL spatial; LOAD spatial; INSTALL httpfs; LOAD httpfs; SET s3_region='us-west-2';")
+    src = f"s3://overturemaps-us-west-2/release/{release}/theme=places/type=place/*"
+    started = time.time()
+    con.sql(f"""
+        COPY (
+          SELECT
+            id,
+            names.primary            AS name,
+            basic_category,
+            taxonomy.primary         AS taxonomy_primary,
+            array_to_string(taxonomy.hierarchy, '>') AS taxonomy_hierarchy,
+            confidence,
+            operating_status,
+            ST_X(geometry)           AS lon,
+            ST_Y(geometry)           AS lat,
+            addresses[1].locality    AS locality,
+            addresses[1].region      AS region,
+            addresses[1].country     AS country
+          FROM read_parquet('{src}', hive_partitioning=1)
+          WHERE bbox.xmin BETWEEN {BBOX['xmin']} AND {BBOX['xmax']}
+            AND bbox.ymin BETWEEN {BBOX['ymin']} AND {BBOX['ymax']}
+        ) TO '{out.as_posix()}' (FORMAT PARQUET)
+    """)
+    n = con.sql(f"SELECT count(*) FROM read_parquet('{out.as_posix()}')").fetchone()[0]
+    print(f"Extracted {n} places to {out} in {time.time() - started:.0f}s")
+    return out
+
+
+if __name__ == "__main__":
+    p = argparse.ArgumentParser()
+    p.add_argument("--release", default=DEFAULT_RELEASE)
+    extract(p.parse_args().release)

intel/ingest/places.py

@@ -0,0 +1,65 @@
+"""
+Loads the local Overture extract (overture_extract.py) into `places`: assigns
+each place to the LGA it falls in (which also drops the Benin/Cameroon/Niger
+points the bounding box caught), maps its category to a subsector, then derives
+locality aliases for the gazetteer.
+
+A release replaces the previous one wholesale — places close and reopen, and
+Overture ids are stable across releases, so a full refresh is simpler and more
+honest than trying to diff.
+"""
+import sqlite3
+
+import duckdb
+
+from intel.freshness import record_attempt
+from intel.geo import PolygonIndex
+from intel.handoff import DATA_STREAM, INTERFACE, produce
+from intel.ingest.overture_extract import DEFAULT_RELEASE, extract_path
+from intel.locations import add_place_localities, lga_polygons
+from intel.subsector_map import subsector_for
+
+SOURCE_ID = "overture_places"
+
+
+def ingest(con: sqlite3.Connection, release: str = DEFAULT_RELEASE, run_id: str = "manual") -> dict:
+    path = extract_path(release)
+    try:
+        if not path.exists():
+            raise FileNotFoundError(f"{path} missing — run `py -m intel.ingest.overture_extract --release {release}` first")
+        rows = duckdb.connect().sql(
+            f"""select id, name, basic_category, taxonomy_primary, taxonomy_hierarchy, confidence,
+                       operating_status, lon, lat, locality
+                from read_parquet('{path.as_posix()}') where lon is not null and lat is not null"""
+        ).fetchall()
+
+        index = PolygonIndex(lga_polygons())
+        con.execute("delete from places where source_id = ?", (SOURCE_ID,))
+        kept = outside = mapped = 0
+        batch = []
+        for pid, name, basic, tax, hier, conf, status, lon, lat, locality in rows:
+            lga = index.lookup(lon, lat)
+            if lga is None:
+                outside += 1
+                continue
+            sub = subsector_for(basic, tax)
+            mapped += sub is not None
+            batch.append((pid, name, basic, hier, sub, conf, status, lat, lon, lga, locality, SOURCE_ID, release))
+            kept += 1
+        con.executemany(
+            """insert into places (id, name, category, taxonomy, subsector, confidence, operating_status,
+                                   lat, lon, location_id, locality, source_id, release)
+               values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+            batch,
+        )
+        con.commit()
+        aliases = add_place_localities(con)
+    except Exception as e:
+        record_attempt(con, SOURCE_ID, ok=False, error=str(e))
+        raise
+
+    record_attempt(con, SOURCE_ID, ok=True, latest_data_period=release.split(".")[0])
+    summary = {"release": release, "places": kept, "outside_nigeria": outside,
+               "with_subsector": mapped, "locality_aliases_added": aliases}
+    produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "places", **summary})
+    return summary

intel/ingest/weather.py

@@ -0,0 +1,178 @@
+"""
+Weather history and forecasts from Open-Meteo, at state centres (Phase 1's
+native level — LGA-level history is the same code with level='lga', paced over
+more days of quota).
+
+Quota is the real constraint, not code: Open-Meteo counts a request covering
+more than 2 weeks or more than 10 variables as several calls (weight ≈ days/14 ×
+max(1, vars/10)), and the free tier allows 600/min, 5,000/hour, 10,000/day. 20
+years × 37 states is ~19k weighted calls, so history is a *resumable backfill*:
+each run fetches whole missing years, oldest first, until it hits its budget,
+and the next run carries on where it stopped. A 429 ends the run cleanly as a
+partial success rather than an error.
+
+The free tier is for non-commercial use — fine for local Phase 1 development; a
+paid plan (or ECMWF open data direct) is needed before this runs for a live
+product (design doc, "Conflicts").
+"""
+import math
+import sqlite3
+import time
+from datetime import date, timedelta
+from typing import Callable
+
+import httpx
+
+from intel.db import now_iso
+from intel.freshness import record_attempt
+from intel.handoff import DATA_STREAM, INTERFACE, produce
+from intel.registry import WEATHER_VARIABLES
+
+ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive"
+FORECAST_URL = "https://api.open-meteo.com/v1/forecast"
+VARIABLES = [v for v, _ in WEATHER_VARIABLES]
+DEFAULT_HISTORY_START = date(2006, 1, 1)
+# The reanalysis archive trails real time by about 5 days.
+ARCHIVE_LAG_DAYS = 6
+PER_MINUTE_BUDGET = 500   # below Open-Meteo's 600/min, leaving headroom
+DEFAULT_RUN_BUDGET = 4500  # below the 5,000/hour limit
+
+Fetch = Callable[[str, dict], dict]
+
+
+class QuotaExhausted(Exception):
+    pass
+
+
+def call_weight(days: int, n_vars: int = len(VARIABLES)) -> float:
+    return max(1.0, days / 14) * max(1.0, n_vars / 10)
+
+
+def http_fetch(url: str, params: dict) -> dict:
+    r = httpx.get(url, params=params, timeout=60.0)
+    if r.status_code == 429:
+        raise QuotaExhausted(r.text[:200])
+    r.raise_for_status()
+    return r.json()
+
+
+def _points(con: sqlite3.Connection, level: str) -> list[sqlite3.Row]:
+    return con.execute("select id, lat, lon from locations where level = ? order by id", (level,)).fetchall()
+
+
+def _store(con: sqlite3.Connection, location_id: str, payload: dict, is_forecast: bool, issued_at: str) -> int:
+    daily = payload["daily"]
+    now = now_iso()
+    rows = []
+    for var in VARIABLES:
+        for day, value in zip(daily["time"], daily.get(var, [])):
+            if value is not None:
+                rows.append((f"weather.{var}", location_id, day, int(is_forecast), issued_at, float(value), now))
+    con.executemany(
+        """insert into observations (stream_id, location_id, period_start, is_forecast, issued_at, value, ingested_at)
+           values (?, ?, ?, ?, ?, ?, ?)
+           on conflict (stream_id, location_id, period_start, is_forecast, issued_at)
+           do update set value = excluded.value, ingested_at = excluded.ingested_at""",
+        rows,
+    )
+    con.commit()
+    return len(rows)
+
+
+def _missing_years(con: sqlite3.Connection, location_id: str, start: date, end: date) -> list[tuple[date, date]]:
+    """Calendar-year spans (clipped to [start, end]) still missing rows for the
+    first variable. A year that's complete from its first day up to some date
+    (the current year, typically) only resumes from the day after; any other
+    gap refetches the whole year — simpler than tracking holes, and rare."""
+    have = {r[0]: (r[1], r[2]) for r in con.execute(
+        """select substr(period_start, 1, 4) y, count(*), max(period_start) from observations
+           where stream_id = ? and location_id = ? and is_forecast = 0 group by y""",
+        (f"weather.{VARIABLES[0]}", location_id),
+    )}
+    spans = []
+    for y in range(start.year, end.year + 1):
+        a, b = max(start, date(y, 1, 1)), min(end, date(y, 12, 31))
+        count, last = have.get(str(y), (0, None))
+        if count >= (b - a).days + 1:
+            continue
+        if last and count == (date.fromisoformat(last) - a).days + 1:
+            a = date.fromisoformat(last) + timedelta(days=1)
+        spans.append((a, b))
+    return spans
+
+
+def backfill_history(con: sqlite3.Connection, level: str = "state", start: date = DEFAULT_HISTORY_START,
+                     budget: float = DEFAULT_RUN_BUDGET, fetch: Fetch = http_fetch,
+                     sleep: Callable[[float], None] = time.sleep, run_id: str = "manual") -> dict:
+    end = date.today() - timedelta(days=ARCHIVE_LAG_DAYS)
+    todo = [(p, a, b) for p in _points(con, level) for a, b in _missing_years(con, p["id"], start, end)]
+    todo.sort(key=lambda t: (t[1], t[0]["id"]))  # oldest years first, across all points
+
+    spent = minute_spent = 0.0
+    minute_started = time.monotonic()
+    fetched = rows = 0
+    stopped = None
+    try:
+        for p, a, b in todo:
+            w = call_weight((b - a).days + 1)
+            if spent + w > budget:
+                stopped = "run budget reached"
+                break
+            if minute_spent + w > PER_MINUTE_BUDGET:
+                sleep(max(0.0, 60 - (time.monotonic() - minute_started)))
+                minute_spent, minute_started = 0.0, time.monotonic()
+            payload = fetch(ARCHIVE_URL, {
+                "latitude": p["lat"], "longitude": p["lon"], "start_date": a.isoformat(), "end_date": b.isoformat(),
+                "daily": ",".join(VARIABLES), "timezone": "Africa/Lagos",
+            })
+            rows += _store(con, p["id"], payload, is_forecast=False, issued_at="")
+            spent += w
+            minute_spent += w
+            fetched += 1
+    except QuotaExhausted as e:
+        stopped = f"Open-Meteo quota: {e}"
+    except Exception as e:
+        record_attempt(con, "open_meteo_archive", ok=False, error=str(e))
+        raise
+
+    newest = con.execute(
+        "select max(period_start) from observations where stream_id like 'weather.%' and is_forecast = 0"
+    ).fetchone()[0]
+    remaining = len(todo) - fetched
+    if fetched or not todo:
+        record_attempt(con, "open_meteo_archive", ok=True, latest_data_period=newest)
+    else:
+        record_attempt(con, "open_meteo_archive", ok=False, error=stopped)
+    summary = {"level": level, "requests": fetched, "weighted_calls": round(spent, 1), "rows": rows,
+               "year_chunks_remaining": remaining, "stopped": stopped, "newest_day": newest}
+    if fetched:
+        produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "weather_history", **summary})
+    return summary
+
+
+def fetch_forecast(con: sqlite3.Connection, level: str = "state", days: int = 16,
+                   fetch: Fetch = http_fetch, run_id: str = "manual") -> dict:
+    issued = date.today().isoformat()
+    rows = n = 0
+    try:
+        for p in _points(con, level):
+            payload = fetch(FORECAST_URL, {
+                "latitude": p["lat"], "longitude": p["lon"], "forecast_days": days,
+                "daily": ",".join(VARIABLES), "timezone": "Africa/Lagos",
+            })
+            rows += _store(con, p["id"], payload, is_forecast=True, issued_at=issued)
+            n += 1
+    except Exception as e:
+        record_attempt(con, "open_meteo_forecast", ok=False, error=str(e))
+        raise
+    record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=issued)
+    summary = {"level": level, "points": n, "rows": rows, "issued_at": issued,
+               "weighted_calls": round(n * call_weight(days), 1)}
+    produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "weather_forecast", **summary})
+    return summary
+
+
+def estimate_backfill_calls(con: sqlite3.Connection, level: str = "state", start: date = DEFAULT_HISTORY_START) -> float:
+    end = date.today() - timedelta(days=ARCHIVE_LAG_DAYS)
+    return math.fsum(call_weight((b - a).days + 1)
+                     for p in _points(con, level) for a, b in _missing_years(con, p["id"], start, end))

intel/locations.py

@@ -0,0 +1,310 @@
+"""
+The gazetteer: country → state → LGA (→ town, later), each with a centre point,
+plus an alias table that turns free text ("flooding in Ajah", "Port Harcourt
+traders") into a location id.
+
+State and LGA *names* come from the app's own list (nigeria_locations.json,
+exported from data/nigeria-locations.ts) so they're exactly what users pick in
+onboarding. *Shapes* come from geoBoundaries, whose LGA spellings differ ("Ado
+Odo/Ota" vs "Ado-Odo/Ota") and whose ADM2 file carries no state, so each
+boundary LGA is first placed in a state by point-in-polygon, then matched to the
+app's name within that state.
+
+resolve() is deliberately simple — exact alias matching on word boundaries,
+with state context used to disambiguate LGA names that exist in several states
+(Surulere, Obi, Bassa, Ifelodun, ...). It's the seed of the Layer 5 search
+service, not a replacement for it.
+"""
+import difflib
+import json
+import re
+import sqlite3
+from pathlib import Path
+
+from intel.freshness import record_attempt
+from intel.geo import PolygonIndex, centroid
+
+HERE = Path(__file__).resolve().parent
+STATES_JSON = HERE / "nigeria_locations.json"
+RAW_DIR = HERE / "data" / "raw"
+ADM1_PATH = RAW_DIR / "geoBoundaries-NGA-ADM1_simplified.geojson"
+ADM2_PATH = RAW_DIR / "geoBoundaries-NGA-ADM2_simplified.geojson"
+
+COUNTRY_ID = "ng"
+FCT_APP_NAME = "FCT (Abuja)"
+ADM1_NAME_TO_APP = {"Abuja Federal Capital Territory": FCT_APP_NAME}
+
+# Boundary-file LGA spellings that normalise differently from the app's name and
+# are too far apart for fuzzy matching to pair safely. Keyed by
+# (app state, normalised boundary name) → app LGA name.
+BOUNDARY_LGA_TO_APP: dict[tuple[str, str], str] = {
+    (FCT_APP_NAME, "municipal area council"): "Abuja Municipal (AMAC)",
+}
+
+# Common place names in news text that aren't state or LGA names. Kept short and
+# only for unambiguous cases; place-data localities add many more automatically.
+CURATED_ALIASES = {
+    "abuja": "ng.fct",
+    "lekki": "ng.lagos.eti-osa",
+    "ajah": "ng.lagos.eti-osa",
+    "victoria island": "ng.lagos.eti-osa",
+    "yaba": "ng.lagos.lagos-mainland",
+    "ibadan": "ng.oyo",
+    "benin city": "ng.edo.oredo",
+    "onitsha": "ng.anambra",
+    "aba": "ng.abia",
+    "warri": "ng.delta",
+    "jos": "ng.plateau",
+    "owerri": "ng.imo",
+    "calabar": "ng.cross-river",
+    "ilorin": "ng.kwara",
+    "enugu": "ng.enugu",
+    "kano": "ng.kano",
+    # National bodies: a policy from these applies to the whole country.
+    "federal government": "ng",
+    "fg": "ng",
+    "cbn": "ng",
+    "central bank of nigeria": "ng",
+    "firs": "ng",
+    "national assembly": "ng",
+    "nigerian": "ng",
+}
+
+
+def normalise(name: str) -> str:
+    s = name.lower().replace("&", " and ")
+    s = re.sub(r"\(.*?\)", " ", s)
+    s = re.sub(r"[^a-z0-9]+", " ", s)
+    return re.sub(r"\s+", " ", s).strip()
+
+
+def _key(name: str) -> str:
+    return normalise(name).replace(" ", "")
+
+
+def slug(name: str) -> str:
+    return normalise(name).replace(" ", "-")
+
+
+def state_id(app_state: str) -> str:
+    return f"{COUNTRY_ID}.fct" if app_state == FCT_APP_NAME else f"{COUNTRY_ID}.{slug(app_state)}"
+
+
+def lga_id(app_state: str, lga: str) -> str:
+    return f"{state_id(app_state)}.{slug(lga)}"
+
+
+def _load_geojson(path: Path) -> list[dict]:
+    return json.loads(path.read_text(encoding="utf8"))["features"]
+
+
+def match_boundaries(states: list[dict], adm1: list[dict], adm2: list[dict]) -> tuple[dict, list[str]]:
+    """Returns ({(app_state, app_lga): adm2 feature}, problems). Every app LGA
+    must end up with exactly one boundary — anything else is reported, not
+    guessed."""
+    app_states = {s["name"] for s in states}
+    state_index = PolygonIndex([
+        (ADM1_NAME_TO_APP.get(f["properties"]["shapeName"], f["properties"]["shapeName"]), f["geometry"]) for f in adm1
+    ])
+    unknown = {k for k, _, _ in state_index._items} - app_states
+    if unknown:
+        raise ValueError(f"Boundary states not in the app list: {sorted(unknown)}")
+
+    by_state: dict[str, list[dict]] = {}
+    problems = []
+    for f in adm2:
+        lon, lat = centroid(f["geometry"])
+        st = state_index.lookup(lon, lat)
+        if st is None:
+            problems.append(f"boundary LGA {f['properties']['shapeName']!r}: centre falls in no state")
+            continue
+        by_state.setdefault(st, []).append(f)
+
+    matched: dict[tuple[str, str], dict] = {}
+    for s in states:
+        st, wanted = s["name"], list(s["lgas"])
+        pool = list(by_state.get(st, []))
+
+        def take(app_lga: str, feature: dict) -> None:
+            matched[(st, app_lga)] = feature
+            wanted.remove(app_lga)
+            pool.remove(feature)
+
+        for f in list(pool):
+            k = _key(f["properties"]["shapeName"])
+            curated = BOUNDARY_LGA_TO_APP.get((st, normalise(f["properties"]["shapeName"])))
+            exact = [w for w in wanted if _key(w) == k]
+            if curated and curated in wanted:
+                take(curated, f)
+            elif len(exact) == 1:
+                take(exact[0], f)
+
+        # Fuzzy pass for what's left, one-to-one, best pairs first.
+        pairs = sorted(
+            ((difflib.SequenceMatcher(None, _key(w), _key(f["properties"]["shapeName"])).ratio(), w, f)
+             for w in wanted for f in pool),
+            key=lambda p: -p[0],
+        )
+        for ratio, w, f in pairs:
+            if ratio < 0.6:
+                break
+            if w in wanted and f in pool:
+                take(w, f)
+
+        for w in wanted:
+            problems.append(f"{st}: app LGA {w!r} has no boundary")
+        for f in pool:
+            problems.append(f"{st}: boundary LGA {f['properties']['shapeName']!r} matched no app LGA")
+    return matched, problems
+
+
+def _add_alias(con: sqlite3.Connection, alias: str, location_id: str, origin: str) -> None:
+    alias = normalise(alias)
+    if alias:
+        con.execute("insert or ignore into location_aliases (alias, location_id, origin) values (?, ?, ?)",
+                    (alias, location_id, origin))
+
+
+def build(con: sqlite3.Connection) -> dict:
+    """(Re)builds country/state/LGA rows and their aliases. Idempotent."""
+    states = json.loads(STATES_JSON.read_text(encoding="utf8"))
+    adm1, adm2 = _load_geojson(ADM1_PATH), _load_geojson(ADM2_PATH)
+    matched, problems = match_boundaries(states, adm1, adm2)
+    if problems:
+        raise ValueError("Boundary matching failed:\n" + "\n".join(problems))
+
+    def upsert(lid, level, name, parent, lon, lat):
+        con.execute(
+            """insert into locations (id, level, name, parent_id, lat, lon, source) values (?, ?, ?, ?, ?, ?, 'geoboundaries')
+               on conflict (id) do update set level = excluded.level, name = excluded.name,
+                 parent_id = excluded.parent_id, lat = excluded.lat, lon = excluded.lon""",
+            (lid, level, name, parent, lat, lon),
+        )
+
+    country = {"type": "MultiPolygon", "coordinates": []}
+    for f in adm1:
+        g = f["geometry"]
+        country["coordinates"] += [g["coordinates"]] if g["type"] == "Polygon" else g["coordinates"]
+    upsert(COUNTRY_ID, "country", "Nigeria", None, *centroid(country))
+    _add_alias(con, "Nigeria", COUNTRY_ID, "gazetteer")
+
+    adm1_by_app = {ADM1_NAME_TO_APP.get(f["properties"]["shapeName"], f["properties"]["shapeName"]): f for f in adm1}
+    for s in states:
+        sid = state_id(s["name"])
+        upsert(sid, "state", s["name"], COUNTRY_ID, *centroid(adm1_by_app[s["name"]]["geometry"]))
+        _add_alias(con, s["name"], sid, "gazetteer")
+        _add_alias(con, f"{s['name']} state", sid, "gazetteer")
+        _add_alias(con, adm1_by_app[s["name"]]["properties"]["shapeName"], sid, "boundary")
+        for lga in s["lgas"]:
+            lid = lga_id(s["name"], lga)
+            f = matched[(s["name"], lga)]
+            upsert(lid, "lga", lga, sid, *centroid(f["geometry"]))
+            _add_alias(con, lga, lid, "gazetteer")
+            _add_alias(con, f["properties"]["shapeName"], lid, "boundary")
+            # "Abuja Municipal (AMAC)" → also findable as "AMAC".
+            for inner in re.findall(r"\((.*?)\)", lga):
+                _add_alias(con, inner, lid, "gazetteer")
+
+    _add_alias(con, "FCT", state_id(FCT_APP_NAME), "gazetteer")
+    for alias, lid in CURATED_ALIASES.items():
+        _add_alias(con, alias, lid, "curated")
+    con.commit()
+    # The boundary files are downloaded by hand (backend/intel/README.md); the
+    # build is what proves they're present and usable, so it reports to Layer 1.
+    # No data period: boundaries have no "newest day", only the fetch clock.
+    record_attempt(con, "geoboundaries", ok=True)
+    n = con.execute("select level, count(*) from locations group by level").fetchall()
+    return {r[0]: r[1] for r in n}
+
+
+def lga_polygons() -> list[tuple[str, dict]]:
+    """(lga id, geometry) for point-in-LGA lookups (places ingest)."""
+    states = json.loads(STATES_JSON.read_text(encoding="utf8"))
+    matched, problems = match_boundaries(states, _load_geojson(ADM1_PATH), _load_geojson(ADM2_PATH))
+    if problems:
+        raise ValueError("Boundary matching failed:\n" + "\n".join(problems))
+    return [(lga_id(st, lga), f["geometry"]) for (st, lga), f in matched.items()]
+
+
+def _alias_rows(con: sqlite3.Connection) -> list[sqlite3.Row]:
+    """alias, location id, its level, and the state it belongs to (itself for a state)."""
+    return con.execute(
+        """select a.alias, l.id, l.level,
+                  case when l.level = 'state' then l.id when p.level = 'state' then p.id end as state
+           from location_aliases a join locations l on l.id = a.location_id
+           left join locations p on p.id = l.parent_id"""
+    ).fetchall()
+
+
+def resolve(con: sqlite3.Connection, text: str | None) -> str | None:
+    """Best location id mentioned in text, or None when there's no mention or
+    the mention is genuinely ambiguous. Prefers the most specific level, with
+    two rules:
+      - a name that is both a state and an LGA ("Bauchi", "Ekiti") means the
+        state — that's how people write;
+      - an LGA name shared by several states ("Surulere") only resolves when the
+        text also names one of those states; otherwise it falls back to their
+        common state if they share one, else None.
+    """
+    if not text:
+        return None
+    hay = f" {normalise(text)} "
+    hits = [r for r in _alias_rows(con) if f" {r['alias']} " in hay]
+    state_aliases = {r["alias"] for r in hits if r["level"] == "state"}
+    hits = [r for r in hits if not (r["level"] in ("lga", "town") and r["alias"] in state_aliases)]
+    named_states = {r["id"] for r in hits if r["level"] == "state"}
+
+    for level in ("town", "lga"):
+        cands = {(r["id"], r["state"]) for r in hits if r["level"] == level}
+        if not cands:
+            continue
+        narrowed = {c for c in cands if c[1] in named_states}
+        cands = narrowed or cands
+        if len(cands) == 1:
+            return next(iter(cands))[0]
+        states = {st for _, st in cands}
+        return next(iter(states)) if len(states) == 1 else None
+
+    if len(named_states) == 1:
+        return next(iter(named_states))
+    if named_states:
+        return None
+    return COUNTRY_ID if any(r["level"] == "country" for r in hits) else None
+
+
+def add_place_localities(con: sqlite3.Connection, min_places: int = 20, share: float = 0.8) -> int:
+    """Turns place-data localities into aliases, by majority vote: a locality
+    string with enough places, most of them inside one LGA, becomes an alias of
+    that LGA; failing that, of their common state. Data-derived, so "Ikorodu" or
+    "Agbara" become resolvable without anyone typing them in."""
+    rows = con.execute(
+        """select lower(trim(locality)) loc, location_id, count(*) n from places
+           where locality is not null and location_id is not null group by 1, 2"""
+    ).fetchall()
+    by_loc: dict[str, list[tuple[str, int]]] = {}
+    for r in rows:
+        by_loc.setdefault(r["loc"], []).append((r["location_id"], r["n"]))
+    added = 0
+    for loc, counts in by_loc.items():
+        total = sum(n for _, n in counts)
+        if total < min_places or not normalise(loc):
+            continue
+        best, n = max(counts, key=lambda c: c[1])
+        if n / total >= share:
+            target = best
+        else:
+            state_counts: dict[str, int] = {}
+            for lid, c in counts:
+                state_counts[lid.rsplit(".", 1)[0]] = state_counts.get(lid.rsplit(".", 1)[0], 0) + c
+            st, sn = max(state_counts.items(), key=lambda c: c[1])
+            if sn / total < share:
+                continue
+            target = st
+        # Never let a derived alias shadow a gazetteer name (e.g. "Lagos" the
+        # locality must keep meaning Lagos State, not whichever LGA dominates).
+        if con.execute("select 1 from location_aliases where alias = ? and origin != 'places'", (normalise(loc),)).fetchone():
+            continue
+        _add_alias(con, loc, target, "places")
+        added += 1
+    con.commit()
+    return added

intel/nigeria_locations.json

@@ -0,0 +1,961 @@
+[
+  {
+    "name": "Abia",
+    "lgas": [
+      "Aba North",
+      "Aba South",
+      "Arochukwu",
+      "Bende",
+      "Ikwuano",
+      "Isiala Ngwa North",
+      "Isiala Ngwa South",
+      "Isuikwuato",
+      "Obi Ngwa",
+      "Ohafia",
+      "Osisioma",
+      "Ugwunagbo",
+      "Ukwa East",
+      "Ukwa West",
+      "Umuahia North",
+      "Umuahia South",
+      "Umu Nneochi"
+    ]
+  },
+  {
+    "name": "Adamawa",
+    "lgas": [
+      "Demsa",
+      "Fufore",
+      "Ganye",
+      "Girei",
+      "Gombi",
+      "Guyuk",
+      "Hong",
+      "Jada",
+      "Lamurde",
+      "Madagali",
+      "Maiha",
+      "Mayo-Belwa",
+      "Michika",
+      "Mubi North",
+      "Mubi South",
+      "Numan",
+      "Shelleng",
+      "Song",
+      "Toungo",
+      "Yola North",
+      "Yola South"
+    ]
+  },
+  {
+    "name": "Akwa Ibom",
+    "lgas": [
+      "Abak",
+      "Eastern Obolo",
+      "Eket",
+      "Esit Eket",
+      "Essien Udim",
+      "Etim Ekpo",
+      "Etinan",
+      "Ibeno",
+      "Ibesikpo Asutan",
+      "Ibiono-Ibom",
+      "Ika",
+      "Ikono",
+      "Ikot Abasi",
+      "Ikot Ekpene",
+      "Ini",
+      "Itu",
+      "Mbo",
+      "Mkpat-Enin",
+      "Nsit-Atai",
+      "Nsit-Ibom",
+      "Nsit-Ubium",
+      "Obot Akara",
+      "Okobo",
+      "Onna",
+      "Oron",
+      "Oruk Anam",
+      "Ukanafun",
+      "Udung-Uko",
+      "Uruan",
+      "Urue-Offong/Oruko",
+      "Uyo"
+    ]
+  },
+  {
+    "name": "Anambra",
+    "lgas": [
+      "Aguata",
+      "Anambra East",
+      "Anambra West",
+      "Anaocha",
+      "Awka North",
+      "Awka South",
+      "Ayamelum",
+      "Dunukofia",
+      "Ekwusigo",
+      "Idemili North",
+      "Idemili South",
+      "Ihiala",
+      "Njikoka",
+      "Nnewi North",
+      "Nnewi South",
+      "Ogbaru",
+      "Onitsha North",
+      "Onitsha South",
+      "Orumba North",
+      "Orumba South",
+      "Oyi"
+    ]
+  },
+  {
+    "name": "Bauchi",
+    "lgas": [
+      "Alkaleri",
+      "Bauchi",
+      "Bogoro",
+      "Dambam",
+      "Darazo",
+      "Dass",
+      "Gamawa",
+      "Ganjuwa",
+      "Giade",
+      "Itas/Gadau",
+      "Jama'are",
+      "Katagum",
+      "Kirfi",
+      "Misau",
+      "Ningi",
+      "Shira",
+      "Tafawa Balewa",
+      "Toro",
+      "Warji",
+      "Zaki"
+    ]
+  },
+  {
+    "name": "Bayelsa",
+    "lgas": [
+      "Brass",
+      "Ekeremor",
+      "Kolokuma/Opokuma",
+      "Nembe",
+      "Ogbia",
+      "Sagbama",
+      "Southern Ijaw",
+      "Yenagoa"
+    ]
+  },
+  {
+    "name": "Benue",
+    "lgas": [
+      "Ado",
+      "Agatu",
+      "Apa",
+      "Buruku",
+      "Gboko",
+      "Guma",
+      "Gwer East",
+      "Gwer West",
+      "Katsina-Ala",
+      "Konshisha",
+      "Kwande",
+      "Logo",
+      "Makurdi",
+      "Obi",
+      "Ogbadibo",
+      "Ohimini",
+      "Oju",
+      "Okpokwu",
+      "Otukpo",
+      "Tarka",
+      "Ukum",
+      "Ushongo",
+      "Vandeikya"
+    ]
+  },
+  {
+    "name": "Borno",
+    "lgas": [
+      "Abadam",
+      "Askira/Uba",
+      "Bama",
+      "Bayo",
+      "Biu",
+      "Chibok",
+      "Damboa",
+      "Dikwa",
+      "Gubio",
+      "Guzamala",
+      "Gwoza",
+      "Hawul",
+      "Jere",
+      "Kaga",
+      "Kala/Balge",
+      "Konduga",
+      "Kukawa",
+      "Kwaya Kusar",
+      "Mafa",
+      "Magumeri",
+      "Maiduguri",
+      "Marte",
+      "Mobbar",
+      "Monguno",
+      "Ngala",
+      "Nganzai",
+      "Shani"
+    ]
+  },
+  {
+    "name": "Cross River",
+    "lgas": [
+      "Abi",
+      "Akamkpa",
+      "Akpabuyo",
+      "Bakassi",
+      "Bekwarra",
+      "Biase",
+      "Boki",
+      "Calabar Municipal",
+      "Calabar South",
+      "Etung",
+      "Ikom",
+      "Obanliku",
+      "Obubra",
+      "Obudu",
+      "Odukpani",
+      "Ogoja",
+      "Yakurr",
+      "Yala"
+    ]
+  },
+  {
+    "name": "Delta",
+    "lgas": [
+      "Aniocha North",
+      "Aniocha South",
+      "Bomadi",
+      "Burutu",
+      "Ethiope East",
+      "Ethiope West",
+      "Ika North East",
+      "Ika South",
+      "Isoko North",
+      "Isoko South",
+      "Ndokwa East",
+      "Ndokwa West",
+      "Okpe",
+      "Oshimili North",
+      "Oshimili South",
+      "Patani",
+      "Sapele",
+      "Udu",
+      "Ughelli North",
+      "Ughelli South",
+      "Ukwuani",
+      "Uvwie",
+      "Warri North",
+      "Warri South",
+      "Warri South West"
+    ]
+  },
+  {
+    "name": "Ebonyi",
+    "lgas": [
+      "Abakaliki",
+      "Afikpo North",
+      "Afikpo South",
+      "Ebonyi",
+      "Ezza North",
+      "Ezza South",
+      "Ikwo",
+      "Ishielu",
+      "Ivo",
+      "Izzi",
+      "Ohaozara",
+      "Ohaukwu",
+      "Onicha"
+    ]
+  },
+  {
+    "name": "Edo",
+    "lgas": [
+      "Akoko-Edo",
+      "Egor",
+      "Esan Central",
+      "Esan North-East",
+      "Esan South-East",
+      "Esan West",
+      "Etsako Central",
+      "Etsako East",
+      "Etsako West",
+      "Igueben",
+      "Ikpoba-Okha",
+      "Orhionmwon",
+      "Oredo",
+      "Ovia North-East",
+      "Ovia South-West",
+      "Owan East",
+      "Owan West",
+      "Uhunmwonde"
+    ]
+  },
+  {
+    "name": "Ekiti",
+    "lgas": [
+      "Ado Ekiti",
+      "Efon",
+      "Ekiti East",
+      "Ekiti South-West",
+      "Ekiti West",
+      "Emure",
+      "Gbonyin",
+      "Ido/Osi",
+      "Ijero",
+      "Ikere",
+      "Ikole",
+      "Ilejemeje",
+      "Irepodun/Ifelodun",
+      "Ise/Orun",
+      "Moba",
+      "Oye"
+    ]
+  },
+  {
+    "name": "Enugu",
+    "lgas": [
+      "Aninri",
+      "Awgu",
+      "Enugu East",
+      "Enugu North",
+      "Enugu South",
+      "Ezeagu",
+      "Igbo Etiti",
+      "Igbo Eze North",
+      "Igbo Eze South",
+      "Isi Uzo",
+      "Nkanu East",
+      "Nkanu West",
+      "Nsukka",
+      "Oji River",
+      "Udenu",
+      "Udi",
+      "Uzo Uwani"
+    ]
+  },
+  {
+    "name": "FCT (Abuja)",
+    "lgas": [
+      "Abaji",
+      "Abuja Municipal (AMAC)",
+      "Bwari",
+      "Gwagwalada",
+      "Kuje",
+      "Kwali"
+    ]
+  },
+  {
+    "name": "Gombe",
+    "lgas": [
+      "Akko",
+      "Balanga",
+      "Billiri",
+      "Dukku",
+      "Funakaye",
+      "Gombe",
+      "Kaltungo",
+      "Kwami",
+      "Nafada",
+      "Shongom",
+      "Yamaltu/Deba"
+    ]
+  },
+  {
+    "name": "Imo",
+    "lgas": [
+      "Aboh Mbaise",
+      "Ahiazu Mbaise",
+      "Ehime Mbano",
+      "Ezinihitte",
+      "Ideato North",
+      "Ideato South",
+      "Ihitte/Uboma",
+      "Ikeduru",
+      "Isiala Mbano",
+      "Isu",
+      "Mbaitoli",
+      "Ngor Okpala",
+      "Njaba",
+      "Nkwerre",
+      "Nwangele",
+      "Obowo",
+      "Oguta",
+      "Ohaji/Egbema",
+      "Okigwe",
+      "Orlu",
+      "Orsu",
+      "Oru East",
+      "Oru West",
+      "Owerri Municipal",
+      "Owerri North",
+      "Owerri West",
+      "Unuimo"
+    ]
+  },
+  {
+    "name": "Jigawa",
+    "lgas": [
+      "Auyo",
+      "Babura",
+      "Birnin Kudu",
+      "Birniwa",
+      "Buji",
+      "Dutse",
+      "Gagarawa",
+      "Garki",
+      "Gumel",
+      "Guri",
+      "Gwaram",
+      "Gwiwa",
+      "Hadejia",
+      "Jahun",
+      "Kafin Hausa",
+      "Kaugama",
+      "Kazaure",
+      "Kiri Kasama",
+      "Kiyawa",
+      "Maigatari",
+      "Malam Madori",
+      "Miga",
+      "Ringim",
+      "Roni",
+      "Sule Tankarkar",
+      "Taura",
+      "Yankwashi"
+    ]
+  },
+  {
+    "name": "Kaduna",
+    "lgas": [
+      "Birnin Gwari",
+      "Chikun",
+      "Giwa",
+      "Igabi",
+      "Ikara",
+      "Jaba",
+      "Jema'a",
+      "Kachia",
+      "Kaduna North",
+      "Kaduna South",
+      "Kagarko",
+      "Kajuru",
+      "Kaura",
+      "Kauru",
+      "Kubau",
+      "Kudan",
+      "Lere",
+      "Makarfi",
+      "Sabon Gari",
+      "Sanga",
+      "Soba",
+      "Zangon Kataf",
+      "Zaria"
+    ]
+  },
+  {
+    "name": "Kano",
+    "lgas": [
+      "Ajingi",
+      "Albasu",
+      "Bagwai",
+      "Bebeji",
+      "Bichi",
+      "Bunkure",
+      "Dala",
+      "Dambatta",
+      "Dawakin Kudu",
+      "Dawakin Tofa",
+      "Doguwa",
+      "Fagge",
+      "Gabasawa",
+      "Garko",
+      "Garun Mallam",
+      "Gaya",
+      "Gezawa",
+      "Gwale",
+      "Gwarzo",
+      "Kabo",
+      "Kano Municipal",
+      "Karaye",
+      "Kibiya",
+      "Kiru",
+      "Kumbotso",
+      "Kunchi",
+      "Kura",
+      "Madobi",
+      "Makoda",
+      "Minjibir",
+      "Nasarawa",
+      "Rano",
+      "Rimin Gado",
+      "Rogo",
+      "Shanono",
+      "Sumaila",
+      "Takai",
+      "Tarauni",
+      "Tofa",
+      "Tsanyawa",
+      "Tudun Wada",
+      "Ungogo",
+      "Warawa",
+      "Wudil"
+    ]
+  },
+  {
+    "name": "Katsina",
+    "lgas": [
+      "Bakori",
+      "Batagarawa",
+      "Batsari",
+      "Baure",
+      "Bindawa",
+      "Charanchi",
+      "Dan Musa",
+      "Dandume",
+      "Danja",
+      "Daura",
+      "Dutsi",
+      "Dutsin-Ma",
+      "Faskari",
+      "Funtua",
+      "Ingawa",
+      "Jibia",
+      "Kafur",
+      "Kaita",
+      "Kankara",
+      "Kankia",
+      "Katsina",
+      "Kurfi",
+      "Kusada",
+      "Mai'Adua",
+      "Malumfashi",
+      "Mani",
+      "Mashi",
+      "Matazu",
+      "Musawa",
+      "Rimi",
+      "Sabuwa",
+      "Safana",
+      "Sandamu",
+      "Zango"
+    ]
+  },
+  {
+    "name": "Kebbi",
+    "lgas": [
+      "Aleiro",
+      "Arewa Dandko",
+      "Argungu",
+      "Augie",
+      "Bagudo",
+      "Birnin Kebbi",
+      "Bunza",
+      "Dandi",
+      "Fakai",
+      "Gwandu",
+      "Jega",
+      "Kalgo",
+      "Koko/Besse",
+      "Maiyama",
+      "Ngaski",
+      "Sakaba",
+      "Shanga",
+      "Suru",
+      "Wasagu/Danko",
+      "Yauri",
+      "Zuru"
+    ]
+  },
+  {
+    "name": "Kogi",
+    "lgas": [
+      "Adavi",
+      "Ajaokuta",
+      "Ankpa",
+      "Bassa",
+      "Dekina",
+      "Ibaji",
+      "Idah",
+      "Igalamela-Odolu",
+      "Ijumu",
+      "Kabba/Bunu",
+      "Kogi",
+      "Lokoja",
+      "Mopa-Muro",
+      "Ofu",
+      "Ogori/Magongo",
+      "Okehi",
+      "Okene",
+      "Olamaboro",
+      "Omala",
+      "Yagba East",
+      "Yagba West"
+    ]
+  },
+  {
+    "name": "Kwara",
+    "lgas": [
+      "Asa",
+      "Baruten",
+      "Edu",
+      "Ekiti",
+      "Ifelodun",
+      "Ilorin East",
+      "Ilorin South",
+      "Ilorin West",
+      "Irepodun",
+      "Isin",
+      "Kaiama",
+      "Moro",
+      "Offa",
+      "Oke Ero",
+      "Oyun",
+      "Pategi"
+    ]
+  },
+  {
+    "name": "Lagos",
+    "lgas": [
+      "Agege",
+      "Ajeromi-Ifelodun",
+      "Alimosho",
+      "Amuwo-Odofin",
+      "Apapa",
+      "Badagry",
+      "Epe",
+      "Eti-Osa",
+      "Ibeju-Lekki",
+      "Ifako-Ijaiye",
+      "Ikeja",
+      "Ikorodu",
+      "Kosofe",
+      "Lagos Island",
+      "Lagos Mainland",
+      "Mushin",
+      "Ojo",
+      "Oshodi-Isolo",
+      "Shomolu",
+      "Surulere"
+    ]
+  },
+  {
+    "name": "Nasarawa",
+    "lgas": [
+      "Akwanga",
+      "Awe",
+      "Doma",
+      "Karu",
+      "Keana",
+      "Keffi",
+      "Kokona",
+      "Lafia",
+      "Nasarawa",
+      "Nasarawa Egon",
+      "Obi",
+      "Toto",
+      "Wamba"
+    ]
+  },
+  {
+    "name": "Niger",
+    "lgas": [
+      "Agaie",
+      "Agwara",
+      "Bida",
+      "Borgu",
+      "Bosso",
+      "Chanchaga",
+      "Edati",
+      "Gbako",
+      "Gurara",
+      "Katcha",
+      "Kontagora",
+      "Lapai",
+      "Lavun",
+      "Magama",
+      "Mariga",
+      "Mashegu",
+      "Mokwa",
+      "Moya",
+      "Paikoro",
+      "Rafi",
+      "Rijau",
+      "Shiroro",
+      "Suleja",
+      "Tafa",
+      "Wushishi"
+    ]
+  },
+  {
+    "name": "Ogun",
+    "lgas": [
+      "Abeokuta North",
+      "Abeokuta South",
+      "Ado-Odo/Ota",
+      "Egbado North",
+      "Egbado South",
+      "Ewekoro",
+      "Ifo",
+      "Ijebu East",
+      "Ijebu North",
+      "Ijebu North East",
+      "Ijebu Ode",
+      "Ikenne",
+      "Imeko Afon",
+      "Ipokia",
+      "Obafemi Owode",
+      "Odeda",
+      "Odogbolu",
+      "Ogun Waterside",
+      "Remo North",
+      "Shagamu"
+    ]
+  },
+  {
+    "name": "Ondo",
+    "lgas": [
+      "Akoko North-East",
+      "Akoko North-West",
+      "Akoko South-East",
+      "Akoko South-West",
+      "Akure North",
+      "Akure South",
+      "Ese Odo",
+      "Idanre",
+      "Ifedore",
+      "Ilaje",
+      "Ile Oluji/Okeigbo",
+      "Irele",
+      "Odigbo",
+      "Okitipupa",
+      "Ondo East",
+      "Ondo West",
+      "Ose",
+      "Owo"
+    ]
+  },
+  {
+    "name": "Osun",
+    "lgas": [
+      "Aiyedaade",
+      "Aiyedire",
+      "Atakunmosa East",
+      "Atakunmosa West",
+      "Boluwaduro",
+      "Boripe",
+      "Ede North",
+      "Ede South",
+      "Egbedore",
+      "Ejigbo",
+      "Ife Central",
+      "Ife East",
+      "Ife North",
+      "Ife South",
+      "Ifedayo",
+      "Ifelodun",
+      "Ila",
+      "Ilesa East",
+      "Ilesa West",
+      "Irepodun",
+      "Irewole",
+      "Isokan",
+      "Iwo",
+      "Obokun",
+      "Odo Otin",
+      "Ola-Oluwa",
+      "Olorunda",
+      "Oriade",
+      "Orolu",
+      "Osogbo"
+    ]
+  },
+  {
+    "name": "Oyo",
+    "lgas": [
+      "Afijio",
+      "Akinyele",
+      "Atiba",
+      "Atisbo",
+      "Egbeda",
+      "Ibadan North",
+      "Ibadan North-East",
+      "Ibadan North-West",
+      "Ibadan South-East",
+      "Ibadan South-West",
+      "Ibarapa Central",
+      "Ibarapa East",
+      "Ibarapa North",
+      "Ido",
+      "Irepo",
+      "Iseyin",
+      "Itesiwaju",
+      "Iwajowa",
+      "Kajola",
+      "Lagelu",
+      "Ogbomosho North",
+      "Ogbomosho South",
+      "Ogo Oluwa",
+      "Olorunsogo",
+      "Oluyole",
+      "Ona Ara",
+      "Orelope",
+      "Orire",
+      "Oyo East",
+      "Oyo West",
+      "Saki East",
+      "Saki West",
+      "Surulere"
+    ]
+  },
+  {
+    "name": "Plateau",
+    "lgas": [
+      "Barkin Ladi",
+      "Bassa",
+      "Bokkos",
+      "Jos East",
+      "Jos North",
+      "Jos South",
+      "Kanam",
+      "Kanke",
+      "Langtang North",
+      "Langtang South",
+      "Mangu",
+      "Mikang",
+      "Pankshin",
+      "Qua'an Pan",
+      "Riyom",
+      "Shendam",
+      "Wase"
+    ]
+  },
+  {
+    "name": "Rivers",
+    "lgas": [
+      "Abua/Odual",
+      "Ahoada East",
+      "Ahoada West",
+      "Akuku-Toru",
+      "Andoni",
+      "Asari-Toru",
+      "Bonny",
+      "Degema",
+      "Eleme",
+      "Emohua",
+      "Etche",
+      "Gokana",
+      "Ikwerre",
+      "Khana",
+      "Obio/Akpor",
+      "Ogba/Egbema/Ndoni",
+      "Ogu/Bolo",
+      "Okrika",
+      "Omuma",
+      "Opobo/Nkoro",
+      "Oyigbo",
+      "Port Harcourt",
+      "Tai"
+    ]
+  },
+  {
+    "name": "Sokoto",
+    "lgas": [
+      "Binji",
+      "Bodinga",
+      "Dange Shuni",
+      "Gada",
+      "Goronyo",
+      "Gudu",
+      "Gwadabawa",
+      "Illela",
+      "Isa",
+      "Kebbe",
+      "Kware",
+      "Rabah",
+      "Sabon Birni",
+      "Shagari",
+      "Silame",
+      "Sokoto North",
+      "Sokoto South",
+      "Tambuwal",
+      "Tangaza",
+      "Tureta",
+      "Wamako",
+      "Wurno",
+      "Yabo"
+    ]
+  },
+  {
+    "name": "Taraba",
+    "lgas": [
+      "Ardo Kola",
+      "Bali",
+      "Donga",
+      "Gashaka",
+      "Gassol",
+      "Ibi",
+      "Jalingo",
+      "Karim Lamido",
+      "Kumi",
+      "Lau",
+      "Sardauna",
+      "Takum",
+      "Ussa",
+      "Wukari",
+      "Yorro",
+      "Zing"
+    ]
+  },
+  {
+    "name": "Yobe",
+    "lgas": [
+      "Bade",
+      "Bursari",
+      "Damaturu",
+      "Fika",
+      "Fune",
+      "Geidam",
+      "Gujba",
+      "Gulani",
+      "Jakusko",
+      "Karasuwa",
+      "Machina",
+      "Nangere",
+      "Nguru",
+      "Potiskum",
+      "Tarmuwa",
+      "Yunusari",
+      "Yusufari"
+    ]
+  },
+  {
+    "name": "Zamfara",
+    "lgas": [
+      "Anka",
+      "Bakura",
+      "Birnin Magaji/Kiyaw",
+      "Bukkuyum",
+      "Bungudu",
+      "Gummi",
+      "Gusau",
+      "Kaura Namoda",
+      "Maradun",
+      "Maru",
+      "Shinkafi",
+      "Talata Mafara",
+      "Tsafe",
+      "Zurmi"
+    ]
+  }
+]

intel/registry.py

@@ -0,0 +1,59 @@
+"""
+The sources and streams the store knows about, defined in code and upserted on
+`init`. Adding a stream is adding an entry here plus an ingester — Layer 1 then
+watches it automatically.
+
+expected_cadence_hours is how often the source publishes, not how often we
+happen to poll it; latency_hours is how far behind "now" its newest data
+legitimately is (Open-Meteo's reanalysis archive trails real time by ~5 days).
+Layer 1 judges freshness against both.
+"""
+import sqlite3
+
+DAY = 24
+SOURCES = [
+    # id, family, provider, licence, expected_cadence_hours, latency_hours
+    ("geoboundaries", "location", "geoBoundaries (William & Mary geoLab)", "CC BY 4.0", 365 * DAY, 0),
+    ("open_meteo_archive", "weather", "Open-Meteo (ECMWF ERA5 reanalysis)", "CC BY 4.0 data; API free tier is non-commercial", DAY, 7 * DAY),
+    ("open_meteo_forecast", "weather", "Open-Meteo (ECMWF IFS and others)", "CC BY 4.0 data; API free tier is non-commercial", DAY, 0),
+    ("overture_places", "places", "Overture Maps Foundation", "CDLA Permissive 2.0", 31 * DAY, 0),
+    ("calendar", "calendar", "VenThrive (computed)", "n/a", 365 * DAY, 0),
+    ("news_pipeline", "policy", "VenThrive news pipeline (ADR-0015)", "per-source", 7 * DAY, 0),
+]
+
+WEATHER_VARIABLES = [
+    # Open-Meteo daily variable, unit
+    ("temperature_2m_max", "°C"),
+    ("temperature_2m_min", "°C"),
+    ("precipitation_sum", "mm"),
+    ("wind_speed_10m_max", "km/h"),
+    # Low humidity with a dry north-easterly is the harmattan signature, so this
+    # is the stream harmattan onset/strength is derived from.
+    ("relative_humidity_2m_mean", "%"),
+]
+
+STREAMS = [
+    # id, family, metric, unit, kind, native_level, cadence, source_id
+    *[(f"weather.{v}", "weather", v, unit, "driver", "state", "daily", "open_meteo_archive") for v, unit in WEATHER_VARIABLES],
+]
+
+
+def seed(con: sqlite3.Connection) -> None:
+    for sid, family, provider, licence, cadence, latency in SOURCES:
+        con.execute(
+            """insert into sources (id, family, provider, licence, expected_cadence_hours, latency_hours)
+               values (?, ?, ?, ?, ?, ?)
+               on conflict (id) do update set family = excluded.family, provider = excluded.provider,
+                 licence = excluded.licence, expected_cadence_hours = excluded.expected_cadence_hours,
+                 latency_hours = excluded.latency_hours""",
+            (sid, family, provider, licence, cadence, latency),
+        )
+    for row in STREAMS:
+        con.execute(
+            """insert into streams (id, family, metric, unit, kind, native_level, cadence, source_id)
+               values (?, ?, ?, ?, ?, ?, ?, ?)
+               on conflict (id) do update set family = excluded.family, metric = excluded.metric,
+                 unit = excluded.unit, kind = excluded.kind, native_level = excluded.native_level,
+                 cadence = excluded.cadence, source_id = excluded.source_id""",
+            row,
+        )

intel/subsector_map.py

@@ -0,0 +1,143 @@
+"""
+Overture place categories → VenThrive subsector ids (backend/news/subsectors.json).
+
+Two levels, most specific first: Overture's detailed `taxonomy.primary` when it
+pins the business down (a 'barber' inside the broad 'personal_or_beauty_service'),
+otherwise its `basic_category`. Anything unmapped gets subsector None — mostly
+context places (worship, civic, parks, landmarks), which are kept because they
+drive footfall even though they aren't competitors.
+
+Built from the categories actually present in Nigeria's extract (release
+2026-09-23.1, 151k places). Known gap: hotels and lodging (~6.8k places) have no
+subsector in data/sectors.ts, so they map to 'other_services' — flagged in the
+design doc as a sector-taxonomy gap rather than forced into the wrong subsector.
+"""
+
+TAXONOMY_TO_SUBSECTOR = {
+    # personal_or_beauty_service, split properly
+    "barber": "hair_services", "hair_salon": "hair_services", "hair_extensions": "hair_services",
+    "hair_replacement": "hair_services", "beauty_salon": "hair_services",
+    "makeup_artist": "skincare_cosmetics", "skin_care_and_makeup": "skincare_cosmetics",
+    "nail_salon": "skincare_cosmetics", "tanning_salon": "body_spa", "hair_removal": "body_spa",
+    "life_coach": "online_coaching", "image_consultant": "management_consulting",
+    "shoe_repair": "footwear", "tattoo_and_piercing": "other_services",
+    # professional_service, split properly
+    "marketing_agency": "digital_services", "advertising_agency": "digital_services",
+    "internet_marketing_service": "digital_services", "social_media_agency": "digital_services",
+    "e_commerce_service": "digital_services", "public_relations": "content_media",
+    "writing_service": "content_media", "employment_agency": "management_consulting",
+    "career_counseling": "management_consulting", "business_consulting": "management_consulting",
+    "janitorial_service": "cleaning_services", "merchandising_service": "retail_trade",
+    # manufacturer, split properly
+    "industrial_equipment_manufacturer": "industrial_manufacturing", "metal_fabricator": "industrial_manufacturing",
+    "auto_manufacturer": "industrial_manufacturing", "motorcycle_manufacturer": "industrial_manufacturing",
+    "plastic_manufacturer": "industrial_manufacturing", "glass_manufacturer": "industrial_manufacturing",
+    "appliance_manufacturer": "light_manufacturing", "furniture_manufacturer": "light_manufacturing",
+    "mattress_manufacturing": "light_manufacturing", "textile_manufacturer": "textiles_materials",
+    "apparel_manufacturer": "clothing_apparel", "shoe_factory_manufacturer": "footwear",
+    "jewelry_and_watches_manufacturer": "accessories_jewellery",
+    # corporate_or_business_office, split properly
+    "information_technology_company": "it_consulting", "telecommunications_company": "it_consulting",
+    "auto_company": "vehicle_trade", "travel_company": "passenger_transport",
+    "industrial_company": "industrial_manufacturing", "pharmaceutical_company": "industrial_manufacturing",
+    "clothing_company": "clothing_apparel", "bottled_water_company": "food_production_processing",
+    "biotechnology_company": "industrial_manufacturing", "ferry_boat_company": "passenger_transport",
+    "bags_luggage_company": "accessories_jewellery",
+}
+
+BASIC_TO_SUBSECTOR = {
+    # food
+    "restaurant": "catering_food_service", "casual_eatery": "catering_food_service",
+    "fast_food_restaurant": "catering_food_service", "cafe": "catering_food_service",
+    "coffee_shop": "catering_food_service", "food_service": "catering_food_service",
+    "food_and_drink": "catering_food_service", "smoothie_juice_bar": "catering_food_service",
+    "food_truck_stand": "catering_food_service", "non_alcoholic_beverage_venue": "catering_food_service",
+    "brewery": "food_production_processing", "winery": "food_production_processing",
+    "distillery": "food_production_processing",
+    "farm": "agricultural_production",
+    "agricultural_service": "agri_trade_distribution", "farmers_market": "agri_trade_distribution",
+    # fashion
+    "fashion_and_apparel_store": "clothing_apparel",
+    # tech
+    "design_service": "digital_services", "technical_service": "hardware_repairs",
+    "b2b_science_and_technology_service": "it_consulting", "telecommunications_service": "it_consulting",
+    # health & beauty
+    "hospital": "healthcare_services", "health_care": "healthcare_services",
+    "diagnostics_imaging_or_lab_service": "healthcare_services", "dental_clinic": "healthcare_services",
+    "outpatient_care_facility": "healthcare_services", "medical_service": "healthcare_services",
+    "vision_or_eye_care_clinic": "healthcare_services", "behavioral_or_mental_health_clinic": "healthcare_services",
+    "physical_medicine_and_rehabilitation": "healthcare_services", "specialized_medical_facility": "healthcare_services",
+    "specialized_health_care": "healthcare_services", "reproductive_perinatal_and_womens_care": "healthcare_services",
+    "primary_care_or_general_clinic": "healthcare_services", "surgery": "healthcare_services",
+    "emergency_department": "healthcare_services", "emergency_or_urgent_care_facility": "healthcare_services",
+    "pediatric_clinic": "healthcare_services", "complementary_and_alternative_medicine": "healthcare_services",
+    "pharmacy_and_drug_store": "healthcare_services", "senior_living_facility": "healthcare_services",
+    "wellness_service": "wellness_fitness", "gym": "wellness_fitness", "fitness_studio": "wellness_fitness",
+    "sport_or_fitness_facility": "wellness_fitness",
+    "personal_or_beauty_service": "hair_services",
+    "personal_care_and_beauty_store": "skincare_cosmetics",
+    # education
+    "place_of_learning": "formal_education", "education": "formal_education",
+    "college_university": "formal_education", "elementary_school": "formal_education",
+    "high_school": "formal_education", "middle_school": "formal_education", "preschool": "formal_education",
+    "educational_facility": "formal_education", "campus_building": "formal_education",
+    "specialty_school": "skills_vocational", "educational_service": "skills_vocational",
+    "tutoring_service": "online_coaching",
+    # money
+    "financial_service": "financial_services", "bank_or_credit_union": "financial_services", "atm": "financial_services",
+    # transport & logistics
+    "shipping_or_delivery_service": "delivery_services",
+    "b2b_transportation_and_storage_service": "haulage_freight", "storage_facility": "haulage_freight",
+    "travel_and_transportation": "passenger_transport", "public_transit_facility_or_service": "passenger_transport",
+    "taxi_or_ride_share_service": "passenger_transport", "travel_service": "passenger_transport",
+    # property & building
+    "real_estate_service": "real_estate", "apartment": "real_estate",
+    "building_or_construction_service": "construction_renovation",
+    "hardware_home_and_garden_store": "building_materials",
+    # creative & events
+    "art_gallery": "visual_arts", "arts_crafts_and_hobby_store": "visual_arts",
+    "media_service": "content_media", "radio_station": "content_media",
+    "theatre_venue": "performing_arts", "performing_arts_venue": "performing_arts",
+    "event_or_party_service": "event_planning", "event_venue": "event_support",
+    "bar": "entertainment_venues", "lounge": "entertainment_venues", "dance_club": "entertainment_venues",
+    "nightlife_venue": "entertainment_venues", "alcoholic_beverage_venue": "entertainment_venues",
+    "music_venue": "entertainment_venues", "movie_theater": "entertainment_venues",
+    "comedy_club": "entertainment_venues", "gaming_venue": "entertainment_venues", "casino": "entertainment_venues",
+    "arcade": "entertainment_venues", "amusement_park": "entertainment_venues", "stadium_arena": "entertainment_venues",
+    # making things
+    "manufacturer": "light_manufacturing", "printing_service": "light_manufacturing",
+    "b2b_industrial_and_machine_service": "industrial_manufacturing",
+    "industrial_facility_or_service": "industrial_manufacturing",
+    # trade
+    "shopping": "retail_trade", "electronics_store": "retail_trade", "food_and_beverage_store": "retail_trade",
+    "convenience_store": "retail_trade", "books_music_and_video_store": "retail_trade",
+    "flowers_and_gifts_store": "retail_trade", "specialty_store": "retail_trade",
+    "sporting_goods_store": "retail_trade", "musical_instrument_and_pro_audio_store": "retail_trade",
+    "office_supply_store": "retail_trade", "second_hand_store": "retail_trade", "department_store": "retail_trade",
+    "discount_store": "retail_trade", "toys_and_games_store": "retail_trade", "superstore": "retail_trade",
+    "shopping_mall": "retail_trade", "warehouse_club_store": "retail_trade", "animal_and_pet_store": "retail_trade",
+    "market": "retail_trade", "gas_station": "retail_trade",
+    "supplier_or_distributor": "wholesale_distribution", "wholesaler": "wholesale_distribution",
+    # professional
+    "attorney_or_law_firm": "legal_compliance", "legal_service": "legal_compliance",
+    # vehicles
+    "automotive_service": "vehicle_maintenance", "vehicle_service": "vehicle_maintenance",
+    "auto_dealer": "vehicle_trade", "vehicle_dealer": "vehicle_trade", "vehicle_parts_store": "vehicle_trade",
+    # services
+    "laundry_service": "cleaning_services", "home_service": "home_services",
+    "hotel": "other_services", "lodging": "other_services", "bed_and_breakfast": "other_services",
+    "private_lodging": "other_services", "inn": "other_services", "resort": "other_services",
+    "professional_service": "other_services", "corporate_or_business_office": "other_services",
+    "b2b_office_and_professional_service": "other_services", "b2b_service": "other_services",
+    "b2b_energy_and_utility_service": "other_services", "security_service": "other_services",
+    "animal_or_pet_service": "other_services", "rental_service": "other_services",
+    "recreational_equipment_rental": "other_services", "environmental_or_ecological_service": "other_services",
+}
+
+
+def subsector_for(basic_category: str | None, taxonomy_primary: str | None) -> str | None:
+    if taxonomy_primary and taxonomy_primary in TAXONOMY_TO_SUBSECTOR:
+        return TAXONOMY_TO_SUBSECTOR[taxonomy_primary]
+    if basic_category:
+        return BASIC_TO_SUBSECTOR.get(basic_category)
+    return None

news/pipeline.py

@@ -22,6 +22,7 @@ from news.classify import classify_item
 from news.relevance import score as relevance_score
 from news.windows import compute_expiry
 from utils.supabase_client import get_supabase
+from intel.events import record_news_item
 
 # Three-way routing by confidence (replaces the old complete/incomplete split).
 # A complete, high-confidence item is business-relevant enough to publish with
@@ -121,7 +122,7 @@ def execute_poll(poll_id: str, since_days: int) -> dict:
                 expires_at = compute_expiry(result.category, scraped_at, result.deadline)
                 status = _route(result)
 
-                db.table("news_items").insert({
+                row = {
                     "source_id":     source["id"],
                     "url":           item.url,
                     "url_hash":      h,
@@ -139,10 +140,15 @@ def execute_poll(poll_id: str, since_days: int) -> dict:
                     "confidence":    result.confidence,
                     "review_reason": result.review_reason(),
                     "classified_at": scraped_at.isoformat(),
-                }).execute()
+                }
+                db.table("news_items").insert(row).execute()
 
                 if status == "classified":
                     counts["items_classified"] += 1
+                    # Published items also become permanent event history for the
+                    # intel store (no-op when it's off) — the feed expires them, the
+                    # correlation step must not lose them.
+                    record_news_item(row)
                 elif status == "rejected":
                     counts["items_rejected"] += 1
                 else:

requirements-intel.txt

@@ -0,0 +1,4 @@
+# Intel store (backend/intel/) — install alongside requirements.txt to run it locally.
+# Not needed by the deployed API: the news hook is a no-op without INTEL_DB_PATH.
+duckdb==1.5.6
+hijridate==2.3.0

routes/news.py

@@ -25,6 +25,7 @@ from news.classify import IMPACTS, CATEGORIES
 from news.business_types import valid_subset
 from utils.supabase_client import get_supabase
 from utils.audit import log_action
+from intel.events import record_news_item
 
 router = APIRouter()
 
@@ -213,4 +214,6 @@ def fix_review_item(item_id: str, body: ReviewFix, user: dict = Depends(verify_f
 
     title = updated[0]["title"] if updated else item_id
     log_action(admin_uid, "review", "published", title, target=item_id)
+    if updated:
+        record_news_item(updated[0])
     return {"success": True, "status": "classified"}

sql/intel_schema.sql

@@ -0,0 +1,136 @@
+-- Intel store: Phase 0 + Phase 1 of the predictive business-intelligence system
+-- (docs/blueprint/intelligence-system.md, ADR-0017..0021).
+--
+-- Portable SQL: runs unchanged on SQLite (Phase 1, local — ADR-0020) and on
+-- Postgres/Supabase when the store moves. The price of that portability:
+-- timestamps are ISO-8601 UTC text, booleans are 0/1 integers, JSON is text, ids
+-- are generated in Python (no serial/uuid defaults). Revisit all three when the
+-- store moves to Postgres for good.
+--
+-- Applied by `py -m intel.cli init` (idempotent — every statement is IF NOT EXISTS).
+
+-- ─── Gazetteer ──────────────────────────────────────────────────────────────
+-- Everything in the store is keyed by location, so this comes first. Ids are
+-- dotted paths: 'ng', 'ng.lagos', 'ng.lagos.ikeja'. State/LGA names are exactly
+-- the ones users pick in onboarding (data/nigeria-locations.ts → nigeria_locations.json).
+create table if not exists locations (
+  id         text primary key,
+  level      text not null check (level in ('country', 'state', 'lga', 'town')),
+  name       text not null,
+  parent_id  text references locations(id),
+  lat        real,
+  lon        real,
+  source     text not null
+);
+
+-- Normalised alternative names → location. Seeded from the gazetteer itself,
+-- boundary-file spellings, a short curated city list, and place-data localities
+-- (intel/locations.py). The seed of the Layer 5 search/geocoding service.
+create table if not exists location_aliases (
+  alias        text not null,
+  location_id  text not null references locations(id),
+  origin       text not null,   -- 'gazetteer' | 'boundary' | 'curated' | 'places'
+  primary key (alias, location_id)
+);
+
+-- ─── Layer 1: stale awareness ───────────────────────────────────────────────
+-- One row per external source. "Stale" means late against the source's OWN
+-- cadence, and two dates are tracked because a fetch can succeed today and still
+-- return last quarter's numbers.
+create table if not exists sources (
+  id                      text primary key,
+  family                  text not null,
+  provider                text not null,
+  licence                 text not null,
+  expected_cadence_hours  integer not null,
+  latency_hours           integer not null default 0,  -- how far behind "now" the newest data legitimately is
+  last_attempt_at         text,
+  last_success_at         text,
+  latest_data_period      text,                         -- newest period the source has given us
+  last_error              text
+);
+
+-- Every layer reports "done" and the receiving layer acknowledges it. Layer 1
+-- flags reports nobody received.
+create table if not exists handoffs (
+  id           text primary key,
+  run_id       text not null,
+  from_layer   text not null,
+  to_layer     text not null,
+  report       text not null,   -- JSON
+  produced_at  text not null,
+  received_at  text,
+  received_by  text
+);
+create index if not exists handoffs_pending on handoffs (to_layer, received_at);
+
+-- ─── Layer 2: data stream ───────────────────────────────────────────────────
+-- A stream is one metric from one source, stored at its NATIVE resolution
+-- (never copied down to finer locations — that would fake precision the models
+-- would then learn from).
+create table if not exists streams (
+  id            text primary key,   -- 'weather.precipitation_sum'
+  family        text not null,
+  metric        text not null,
+  unit          text not null,
+  kind          text not null check (kind in ('driver', 'indicator')),
+  native_level  text not null check (native_level in ('country', 'state', 'lga', 'town', 'point')),
+  cadence       text not null check (cadence in ('daily', 'weekly', 'monthly', 'quarterly', 'yearly', 'event')),
+  source_id     text not null references sources(id)
+);
+
+-- Numeric time series. Observed history and forecasts share one table:
+-- history has is_forecast=0 and issued_at=''; each forecast run keeps its own
+-- rows (issued_at = run date) so forecast skill can be measured later.
+create table if not exists observations (
+  stream_id     text not null references streams(id),
+  location_id   text not null references locations(id),
+  period_start  text not null,
+  is_forecast   integer not null default 0 check (is_forecast in (0, 1)),
+  issued_at     text not null default '',
+  value         real not null,
+  ingested_at   text not null,
+  primary key (stream_id, location_id, period_start, is_forecast, issued_at)
+);
+create index if not exists observations_loc on observations (location_id, stream_id, period_start);
+
+-- Things that happen rather than values that move: a policy enacted, a holiday,
+-- an incident, a project gazetted. Kept permanently — this is the history the
+-- correlation step and models learn from (unlike news_items, which expire from
+-- the user-facing feed).
+create table if not exists events (
+  id           text primary key,
+  family       text not null,   -- policy | weather | infrastructure | funding | security | calendar | other
+  type         text not null,   -- e.g. 'public_holiday', 'religious_holiday', 'season', 'news'
+  location_id  text references locations(id),
+  starts_on    text not null,
+  ends_on      text,
+  title        text not null,
+  detail       text,
+  is_estimate  integer not null default 0 check (is_estimate in (0, 1)),
+  source_id    text not null references sources(id),
+  source_ref   text not null,   -- natural key at the source, so re-ingest is idempotent
+  ingested_at  text not null,
+  unique (source_id, source_ref)
+);
+create index if not exists events_loc on events (location_id, starts_on);
+
+-- Points of interest. subsector is one of backend/news/subsectors.json, or null
+-- for a context place (worship, civic, landmark) — those still matter as
+-- footfall generators even though they aren't competitors.
+create table if not exists places (
+  id                text primary key,   -- Overture GERS id
+  name              text,
+  category          text,
+  taxonomy          text,
+  subsector         text,
+  confidence        real,
+  operating_status  text,
+  lat               real not null,
+  lon               real not null,
+  location_id       text references locations(id),   -- the LGA it falls in
+  locality          text,
+  source_id         text not null references sources(id),
+  release           text not null
+);
+create index if not exists places_loc on places (location_id, subsector);

tests/test_intel.py

@@ -0,0 +1,319 @@
+"""
+Unit tests for the intel store (Phase 0 + 1, docs/blueprint/intelligence-system.md):
+geometry, gazetteer matching and text resolution, Layer 1 freshness, handoff
+receipts, the calendar, resumable weather backfill, category mapping, and the
+news → events hook. Everything runs on in-memory SQLite with synthetic data —
+no network, and no dependency on the gitignored raw downloads.
+"""
+from datetime import date, datetime, timedelta, timezone
+
+import pytest
+
+from intel import db, handoff
+from intel.freshness import record_attempt, report, source_status
+from intel.geo import PolygonIndex, centroid, contains, haversine_km
+from intel.ingest import calendar as cal
+from intel.ingest import weather
+from intel.locations import match_boundaries, normalise, resolve
+from intel.subsector_map import subsector_for
+
+
+def square(x0, y0, x1, y1):
+    return {"type": "Polygon", "coordinates": [[[x0, y0], [x1, y0], [x1, y1], [x0, y1], [x0, y0]]]}
+
+
+@pytest.fixture
+def con():
+    c = db.connect(":memory:")
+    db.init(c)
+    yield c
+    c.close()
+
+
+def add_location(con, lid, level, name, parent=None, lat=0.0, lon=0.0, aliases=()):
+    con.execute("insert into locations (id, level, name, parent_id, lat, lon, source) values (?, ?, ?, ?, ?, ?, 'test')",
+                (lid, level, name, parent, lat, lon))
+    for a in (name, *aliases):
+        con.execute("insert into location_aliases (alias, location_id, origin) values (?, ?, 'test')", (normalise(a), lid))
+
+
+@pytest.fixture
+def gazetteer(con):
+    add_location(con, "ng", "country", "Nigeria")
+    add_location(con, "ng.lagos", "state", "Lagos", "ng")
+    add_location(con, "ng.oyo", "state", "Oyo", "ng")
+    add_location(con, "ng.bauchi", "state", "Bauchi", "ng")
+    add_location(con, "ng.lagos.surulere", "lga", "Surulere", "ng.lagos")
+    add_location(con, "ng.oyo.surulere", "lga", "Surulere", "ng.oyo")
+    add_location(con, "ng.lagos.eti-osa", "lga", "Eti-Osa", "ng.lagos", aliases=("Ajah", "Lekki"))
+    add_location(con, "ng.bauchi.bauchi", "lga", "Bauchi", "ng.bauchi")
+    con.commit()
+    return con
+
+
+# ─── geometry ───────────────────────────────────────────────────────────────
+
+def test_centroid_of_square():
+    assert centroid(square(0, 0, 2, 2)) == pytest.approx((1.0, 1.0))
+
+
+def test_centroid_respects_holes():
+    g = {"type": "Polygon", "coordinates": [
+        [[0, 0], [4, 0], [4, 4], [0, 4], [0, 0]],
+        [[2, 0], [4, 0], [4, 4], [2, 4], [2, 0]],  # right half removed
+    ]}
+    assert centroid(g)[0] == pytest.approx(1.0)
+
+
+def test_contains_and_index():
+    idx = PolygonIndex([("a", square(0, 0, 1, 1)), ("b", square(1, 0, 2, 1))])
+    assert idx.lookup(0.5, 0.5) == "a"
+    assert idx.lookup(1.5, 0.5) == "b"
+    assert idx.lookup(5, 5) is None
+    assert not contains(square(0, 0, 1, 1), 1.5, 0.5)
+
+
+def test_haversine_one_degree_at_equator():
+    assert haversine_km(0, 0, 1, 0) == pytest.approx(111.2, abs=0.2)
+
+
+# ─── gazetteer ──────────────────────────────────────────────────────────────
+
+def _feature(name, geom):
+    return {"properties": {"shapeName": name}, "geometry": geom}
+
+
+def test_match_boundaries_exact_fuzzy_and_state_placement():
+    states = [{"name": "Lagos", "lgas": ["Ifako-Ijaiye", "Surulere"]}, {"name": "Oyo", "lgas": ["Surulere"]}]
+    adm1 = [_feature("Lagos", square(0, 0, 2, 1)), _feature("Oyo", square(0, 1, 2, 2))]
+    adm2 = [
+        _feature("Ifako/Ijaye", square(0, 0, 1, 1)),   # fuzzy
+        _feature("Surulere", square(1, 0, 2, 1)),      # exact, in Lagos by position
+        _feature("Surulere", square(0, 1, 2, 2)),      # same name, in Oyo by position
+    ]
+    matched, problems = match_boundaries(states, adm1, adm2)
+    assert problems == []
+    assert matched[("Lagos", "Ifako-Ijaiye")]["properties"]["shapeName"] == "Ifako/Ijaye"
+    assert centroid(matched[("Oyo", "Surulere")]["geometry"])[1] == pytest.approx(1.5)
+
+
+def test_match_boundaries_reports_rather_than_guesses():
+    states = [{"name": "Lagos", "lgas": ["Agege"]}]
+    adm1 = [_feature("Lagos", square(0, 0, 1, 1))]
+    adm2 = [_feature("Completely Different", square(0, 0, 1, 1))]
+    _, problems = match_boundaries(states, adm1, adm2)
+    assert any("Agege" in p for p in problems)
+
+
+def test_resolve_alias_to_lga(gazetteer):
+    assert resolve(gazetteer, "Flooding in Ajah after heavy rain") == "ng.lagos.eti-osa"
+
+
+def test_resolve_shared_lga_name_needs_state_context(gazetteer):
+    assert resolve(gazetteer, "Traders in Surulere protest") is None
+    assert resolve(gazetteer, "Surulere, Lagos market shut") == "ng.lagos.surulere"
+
+
+def test_resolve_state_beats_same_named_lga(gazetteer):
+    assert resolve(gazetteer, "Bauchi governor signs levy") == "ng.bauchi"
+
+
+def test_resolve_no_mention(gazetteer):
+    assert resolve(gazetteer, "Celebrity wedding photos") is None
+    assert resolve(gazetteer, None) is None
+
+
+def test_resolve_word_boundaries(gazetteer):
+    # "Oyo" must not match inside another word.
+    assert resolve(gazetteer, "Coyotes spotted") is None
+
+
+# ─── Layer 1: freshness ─────────────────────────────────────────────────────
+
+def _source(con, sid):
+    return con.execute("select * from sources where id = ?", (sid,)).fetchone()
+
+
+def test_never_fetched_source_is_never(con):
+    assert source_status(_source(con, "calendar"), datetime.now(timezone.utc))["status"] == "never"
+
+
+def test_fresh_then_stale_on_fetch_clock(con):
+    record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=date.today().isoformat())
+    now = datetime.now(timezone.utc)
+    assert source_status(_source(con, "open_meteo_forecast"), now)["status"] == "fresh"
+    assert source_status(_source(con, "open_meteo_forecast"), now + timedelta(days=2))["status"] == "late"
+    assert source_status(_source(con, "open_meteo_forecast"), now + timedelta(days=4))["status"] == "stale"
+
+
+def test_data_clock_catches_old_data_behind_a_fresh_fetch(con):
+    # Fetched just now, but the newest data is a month old on a daily source.
+    old = (date.today() - timedelta(days=30)).isoformat()
+    record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=old)
+    s = source_status(_source(con, "open_meteo_forecast"), datetime.now(timezone.utc))
+    assert s["status"] == "stale"
+    assert "newest data" in s["reason"]
+
+
+def test_archive_latency_is_allowed(con):
+    # The archive legitimately trails by ~a week; 6-day-old data is on time.
+    record_attempt(con, "open_meteo_archive", ok=True, latest_data_period=(date.today() - timedelta(days=6)).isoformat())
+    assert source_status(_source(con, "open_meteo_archive"), datetime.now(timezone.utc))["status"] == "fresh"
+
+
+def test_failure_after_success_is_failing(con):
+    record_attempt(con, "open_meteo_forecast", ok=True, latest_data_period=date.today().isoformat())
+    record_attempt(con, "open_meteo_forecast", ok=False, error="timeout")
+    s = source_status(_source(con, "open_meteo_forecast"), datetime.now(timezone.utc))
+    assert s["status"] == "failing" and "timeout" in s["reason"]
+
+
+# ─── handoffs ───────────────────────────────────────────────────────────────
+
+def test_handoff_receipt_once(con):
+    hid = handoff.produce(con, "r1", handoff.DATA_STREAM, handoff.STALE_AWARENESS, {"ok": True})
+    assert [h["id"] for h in handoff.pending(con, handoff.STALE_AWARENESS)] == [hid]
+    assert handoff.receive(con, hid, "test") is True
+    assert handoff.receive(con, hid, "test") is False
+    assert handoff.pending(con, handoff.STALE_AWARENESS) == []
+
+
+def test_unreceived_handoff_to_active_layer_goes_overdue(con):
+    handoff.produce(con, "r1", handoff.DATA_STREAM, handoff.STALE_AWARENESS, {})
+    later = datetime.now(timezone.utc) + timedelta(days=2)
+    assert len(report(con, later)["handoffs_overdue"]) == 1
+
+
+def test_handoff_to_unbuilt_layer_is_queued_not_overdue(con):
+    handoff.produce(con, "r1", handoff.DATA_STREAM, handoff.INTERFACE, {})
+    r = report(con, datetime.now(timezone.utc) + timedelta(days=30))
+    assert r["handoffs_overdue"] == [] and r["handoffs_queued_for_unbuilt_layers"] == 1
+
+
+# ─── calendar ───────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("year,expected", [(2024, date(2024, 3, 31)), (2025, date(2025, 4, 20)), (2026, date(2026, 4, 5))])
+def test_easter(year, expected):
+    assert cal.easter(year) == expected
+
+
+def test_democracy_day_moved_in_2019():
+    dd = lambda y: next(e["starts_on"] for e in cal.events_for_year(y) if e["key"] == "democracy_day")  # noqa: E731
+    assert dd(2018) == date(2018, 5, 29)
+    assert dd(2019) == date(2019, 6, 12)
+
+
+def test_islamic_holidays_are_estimates():
+    events = {e["key"]: e for e in cal.events_for_year(2025)}
+    assert events["eid_al_fitr"]["starts_on"] == date(2025, 3, 30)
+    assert events["eid_al_fitr"]["is_estimate"] and not events["christmas"]["is_estimate"]
+
+
+def test_calendar_ingest_is_idempotent(gazetteer):
+    first = cal.ingest(gazetteer, 2024, 2025)["events"]
+    cal.ingest(gazetteer, 2024, 2025)
+    assert gazetteer.execute("select count(*) from events where source_id = 'calendar'").fetchone()[0] <= first
+
+
+# ─── weather backfill ───────────────────────────────────────────────────────
+
+def fake_fetch(calls):
+    def fetch(url, params):
+        calls.append(params)
+        a, b = date.fromisoformat(params.get("start_date") or date.today().isoformat()), \
+            date.fromisoformat(params.get("end_date") or (date.today() + timedelta(days=params.get("forecast_days", 1) - 1)).isoformat())
+        days = [(a + timedelta(days=i)).isoformat() for i in range((b - a).days + 1)]
+        return {"daily": {"time": days, **{v: [1.0] * len(days) for v in weather.VARIABLES}}}
+    return fetch
+
+
+def test_call_weight():
+    assert weather.call_weight(14) == 1
+    assert weather.call_weight(365) == pytest.approx(365 / 14)
+
+
+def test_backfill_stops_at_budget_and_resumes(gazetteer):
+    start = date(date.today().year - 3, 1, 1)
+    calls = []
+    first = weather.backfill_history(gazetteer, start=start, budget=60, fetch=fake_fetch(calls), sleep=lambda s: None)
+    assert first["stopped"] == "run budget reached" and first["year_chunks_remaining"] > 0
+    weather.backfill_history(gazetteer, start=start, budget=10_000, fetch=fake_fetch(calls), sleep=lambda s: None)
+    assert weather._missing_years(gazetteer, "ng.lagos", start, date.today() - timedelta(days=weather.ARCHIVE_LAG_DAYS)) == []
+    # A third run has nothing to fetch.
+    n = len(calls)
+    weather.backfill_history(gazetteer, start=start, budget=10_000, fetch=fake_fetch(calls), sleep=lambda s: None)
+    assert len(calls) == n
+
+
+def test_backfill_resumes_current_year_from_last_day(gazetteer):
+    end = date.today() - timedelta(days=weather.ARCHIVE_LAG_DAYS)
+    start = date(end.year, 1, 1)
+    weather.backfill_history(gazetteer, start=start, fetch=fake_fetch([]), sleep=lambda s: None)
+    gazetteer.execute("delete from observations where period_start > ?", ((end - timedelta(days=3)).isoformat(),))
+    spans = weather._missing_years(gazetteer, "ng.lagos", start, end)
+    assert spans == [(end - timedelta(days=2), end)]
+
+
+def test_quota_429_is_partial_success_not_error(gazetteer):
+    def fetch(url, params):
+        raise weather.QuotaExhausted("daily limit")
+    out = weather.backfill_history(gazetteer, start=date(date.today().year, 1, 1), fetch=fetch, sleep=lambda s: None)
+    assert "quota" in out["stopped"]
+
+
+def test_forecast_rows_are_marked_and_versioned(gazetteer):
+    weather.fetch_forecast(gazetteer, days=3, fetch=fake_fetch([]))
+    rows = gazetteer.execute("select distinct is_forecast, issued_at from observations").fetchall()
+    assert [tuple(r) for r in rows] == [(1, date.today().isoformat())]
+
+
+# ─── places category mapping ────────────────────────────────────────────────
+
+def test_subsector_prefers_detailed_taxonomy():
+    assert subsector_for("personal_or_beauty_service", "barber") == "hair_services"
+    assert subsector_for("personal_or_beauty_service", "makeup_artist") == "skincare_cosmetics"
+    assert subsector_for("restaurant", None) == "catering_food_service"
+
+
+def test_context_places_have_no_subsector():
+    assert subsector_for("christian_place_of_worship", None) is None
+    assert subsector_for(None, None) is None
+
+
+def test_mapped_subsectors_are_real_ids():
+    from news.business_types import VALID_IDS
+    from intel.subsector_map import BASIC_TO_SUBSECTOR, TAXONOMY_TO_SUBSECTOR
+    assert set(BASIC_TO_SUBSECTOR.values()) | set(TAXONOMY_TO_SUBSECTOR.values()) <= set(VALID_IDS)
+
+
+# ─── news → events ──────────────────────────────────────────────────────────
+
+def test_record_news_item_is_noop_when_store_off(monkeypatch):
+    from intel.events import record_news_item
+    monkeypatch.delenv("INTEL_DB_PATH", raising=False)
+    assert record_news_item({"url_hash": "x", "title": "t"}) is False
+
+
+def test_record_news_item_writes_located_event(monkeypatch, tmp_path):
+    from intel.events import record_news_item
+    path = tmp_path / "intel.db"
+    monkeypatch.setenv("INTEL_DB_PATH", str(path))
+    c = db.connect()
+    db.init(c)
+    add_location(c, "ng", "country", "Nigeria")
+    add_location(c, "ng.lagos", "state", "Lagos", "ng")
+    add_location(c, "ng.lagos.eti-osa", "lga", "Eti-Osa", "ng.lagos", aliases=("Lekki",))
+    c.commit()
+    item = {"url_hash": "abc", "title": "Lekki traders face new levy", "category": "policy", "location": "Lekki",
+            "published_at": "2026-10-01T09:00:00+00:00", "raw_text": "..."}
+    assert record_news_item(item) is True
+    assert record_news_item(item) is True  # idempotent upsert
+    rows = c.execute("select family, location_id, starts_on from events where source_id = 'news_pipeline'").fetchall()
+    assert [tuple(r) for r in rows] == [("policy", "ng.lagos.eti-osa", "2026-10-01")]
+    c.close()
+
+
+def test_record_news_item_never_raises(monkeypatch, tmp_path):
+    from intel.events import record_news_item
+    monkeypatch.setenv("INTEL_DB_PATH", str(tmp_path / "missing-schema.db"))
+    assert record_news_item({"url_hash": "x", "title": "t"}) is False