navi-offroute: HPA tile DB manifest pattern (H6a-pre, unblocks continental rollout) (#53)

* navi-offroute: HPA tile DB manifest pattern (H6a-pre, unblocks continental rollout)

Replaces the single-file NAVI_OFFROUTE_HPA_DB env var with a directory-based
manifest layout (NAVI_OFFROUTE_HPA_DIR + manifest.json) so one deployment can
carry multiple regional tile DBs side by side and look up which one(s) cover
a route bbox. The two-level HPA* kernel (astar_hpa_multimode) is unchanged;
router.py asks the new hpa_manifest module for matching tile DBs and forwards
the path to the existing kernel.

Manifest schema (version 1):
  {"version": 1, "tile_dbs": [
    {"name": "idaho", "path": "idaho.db",
     "chunk_bounds": {"min_x": -8550, "max_x": -8430, "min_y": 3100, "max_y": 3260}}
  ]}

- path is relative to the manifest directory.
- chunk_bounds is inclusive in chunk-index space; null = covers everywhere.
- Multiple entries may overlap; v1 dispatch only handles single-region routes,
  multi-region falls through to the unified kernel (UNION across DBs is a
  future PR — needs astar_hpa_multimode signature change).

Backward-compat: if NAVI_OFFROUTE_HPA_DB is set and NAVI_OFFROUTE_HPA_DIR is
not, the loader synthesizes a single unbounded entry and warns (deprecation).
Current deploy keeps working byte-identically; env-file migration happens as
a separate follow-up step.

Touches:
  + hpa_manifest.py (new, 144 LOC) — schema, loaders, lookup, lazy conn cache
  ~ router.py (-24 +31) — _HPA_MANIFEST replaces HPA_TILE_DB; dispatch site
    consults manifest.dbs_for_route_bbox; _hpa_eligible uses manifest.enabled
  + tests/test_hpa_manifest.py (new, 117 LOC) — 8 tests covering all paths
  ~ tests/test_hpa_runtime.py (4 lines) — monkeypatch _HPA_MANIFEST instead

No touches to hpa_build, astar, transitions, cost, mvum. No schema change.
No service restart, no env-file edit (deploy migration happens after merge).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* navi-offroute: drop unused connection cache from HPAManifest (cleanup)

The connection cache field (_conns) and get_connection() method were
shipped in PR #53 "ready for future use" but never consumed by the v1
dispatch path: astar_hpa_multimode opens its own sqlite connection per
call (astar.py:752) and changing that signature was out of scope.

Per self-review of PR #53: shipping dead code in a PR is wrong even
when the rationale is "future PRs will use it." Removing it here keeps
the manifest module focused on lookup; whoever needs the cache later
can add it alongside the use site.

  - Drop _conns field, get_connection() method.
  - Drop sqlite3 + Dict imports (no longer referenced).
  - Drop test_get_connection_caches_per_path.

23 deletions, 2 insertions. No production behaviour change.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Ubuntu <zvx@recon-vm.echo6.co>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/services/navi_offroute/hpa_manifest.py

@@ -0,0 +1,130 @@
+"""HPA tile-DB manifest pattern (H6a-pre). Replaces the single-file
+``NAVI_OFFROUTE_HPA_DB`` env var with a directory-based manifest
+(``NAVI_OFFROUTE_HPA_DIR``) so one deployment can carry multiple regional tile
+DBs and look up which one(s) cover a route bbox. The two-level HPA* kernel
+(``astar_hpa_multimode``) is unchanged; ``router.py`` asks this module for
+matching tile DBs and forwards the path. Schema is encoded in the loaders
+below; backward-compat: legacy ``HPA_DB`` alone -> single unbounded entry,
+neither env var set -> HPA disabled (same as today).
+"""
+import json
+import logging
+import os
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+logger = logging.getLogger(__name__)
+
+MANIFEST_VERSION = 1
+ENV_DIR = "NAVI_OFFROUTE_HPA_DIR"
+ENV_LEGACY = "NAVI_OFFROUTE_HPA_DB"
+
+
+@dataclass(frozen=True)
+class ChunkBounds:
+    """Inclusive chunk-index rectangle."""
+    min_x: int
+    max_x: int
+    min_y: int
+    max_y: int
+
+    def intersects(self, cx_min: int, cx_max: int, cy_min: int, cy_max: int) -> bool:
+        return not (cx_max < self.min_x or cx_min > self.max_x
+                    or cy_max < self.min_y or cy_min > self.max_y)
+
+
+@dataclass(frozen=True)
+class TileDBEntry:
+    name: str
+    abs_path: str
+    chunk_bounds: Optional[ChunkBounds]   # None -> unbounded
+
+
+@dataclass
+class HPAManifest:
+    """Registry of regional tile DBs. Immutable after load."""
+    entries: List[TileDBEntry] = field(default_factory=list)
+
+    def enabled(self) -> bool:
+        return len(self.entries) > 0
+
+    def dbs_for_chunks(self, cx_min: int, cx_max: int,
+                       cy_min: int, cy_max: int) -> List[str]:
+        """Tile DBs whose chunk_bounds intersect the chunk rectangle. Unbounded
+        entries always match. Preserves declaration order."""
+        out = []
+        for e in self.entries:
+            if e.chunk_bounds is None or e.chunk_bounds.intersects(
+                    cx_min, cx_max, cy_min, cy_max):
+                out.append(e.abs_path)
+        return out
+
+    def dbs_for_route_bbox(self, south: float, north: float,
+                           west: float, east: float) -> List[str]:
+        """Resolve a route's degree bbox to its chunk-index rectangle and look up
+        matching DB paths. Local import keeps this module independent of the
+        build pipeline's import graph."""
+        from .hpa_build import chunk_coords
+        cx0, cy0 = chunk_coords(south, west)
+        cx1, cy1 = chunk_coords(north, east)
+        return self.dbs_for_chunks(min(cx0, cx1), max(cx0, cx1),
+                                   min(cy0, cy1), max(cy0, cy1))
+
+# ── loaders ──────────────────────────────────────────────────────────────────
+
+def _load_from_dir(dir_path: str) -> HPAManifest:
+    """Parse ``<dir>/manifest.json``. Fails fast on bad version, missing file,
+    or any entry whose resolved path is absent."""
+    mpath = os.path.join(dir_path, "manifest.json")
+    if not os.path.isfile(mpath):
+        raise FileNotFoundError(f"HPA manifest not found at {mpath}")
+    with open(mpath) as f:
+        data = json.load(f)
+    if data.get("version") != MANIFEST_VERSION:
+        raise ValueError(
+            f"HPA manifest at {mpath}: unsupported version "
+            f"{data.get('version')!r} (expected {MANIFEST_VERSION})")
+    entries: List[TileDBEntry] = []
+    for raw in data.get("tile_dbs") or []:
+        name = raw["name"]
+        rel = raw["path"]
+        abs_path = rel if os.path.isabs(rel) else os.path.join(dir_path, rel)
+        if not os.path.isfile(abs_path):
+            raise FileNotFoundError(
+                f"HPA manifest at {mpath}: entry {name!r} -> {abs_path} not found")
+        cb_raw = raw.get("chunk_bounds")
+        cb = (ChunkBounds(min_x=int(cb_raw["min_x"]), max_x=int(cb_raw["max_x"]),
+                          min_y=int(cb_raw["min_y"]), max_y=int(cb_raw["max_y"]))
+              if cb_raw is not None else None)
+        entries.append(TileDBEntry(name=name, abs_path=abs_path, chunk_bounds=cb))
+    return HPAManifest(entries=entries)
+
+
+def _load_from_legacy(legacy_path: str) -> HPAManifest:
+    """Synthesize a single unbounded entry from ``NAVI_OFFROUTE_HPA_DB``. Always
+    warns; missing file -> empty (HPA disabled, same as today)."""
+    if not os.path.isfile(legacy_path):
+        logger.warning("%s=%s does not exist; HPA disabled", ENV_LEGACY, legacy_path)
+        return HPAManifest(entries=[])
+    logger.warning(
+        "%s is deprecated; migrate to %s + manifest.json. "
+        "Treating %s as an unbounded single-entry manifest.",
+        ENV_LEGACY, ENV_DIR, legacy_path)
+    return HPAManifest(entries=[TileDBEntry(
+        name="legacy", abs_path=legacy_path, chunk_bounds=None)])
+
+
+def load() -> HPAManifest:
+    """Load the active manifest from env. ``HPA_DIR`` wins over ``HPA_DB`` if
+    both are set (warns); neither -> empty (HPA disabled)."""
+    dir_env = os.environ.get(ENV_DIR)
+    legacy_env = os.environ.get(ENV_LEGACY)
+    if dir_env:
+        if legacy_env:
+            logger.warning(
+                "Both %s and %s are set; %s takes precedence",
+                ENV_DIR, ENV_LEGACY, ENV_DIR)
+        return _load_from_dir(dir_env)
+    if legacy_env:
+        return _load_from_legacy(legacy_env)
+    return HPAManifest(entries=[])

backend/services/navi_offroute/router.py

@@ -48,6 +48,7 @@ from .trails import TrailReader
 from .mvum import get_mvum_access_grid, get_mvum_access_grids_all_modes
 from .mvum_annotate import annotate_network_edges
 from .mvum_exclude import build_exclude_polygons
+from . import hpa_manifest
 
 logger = logging.getLogger("navi_offroute.router")
 
@@ -61,10 +62,11 @@ POSTGIS_DSN = os.environ.get("NAVI_OFFROUTE_POSTGIS_DSN", "dbname=padus")
 # Valhalla endpoint (recon-side network router, HTTP)
 VALHALLA_URL = os.environ.get("NAVI_OFFROUTE_VALHALLA_URL", "http://localhost:8002")
 
-# HPA* cost-tile DB (HPA-SPEC.md §8/§9, Phase H3). Unset (None) -> HPA* never engages and
-# Auto routing is byte-identical to the unified-graph path. Set to a tile DB (built by
-# hpa_build) to enable the two-level fast path for covered, pragmatic, no-affinity routes.
-HPA_TILE_DB = os.environ.get("NAVI_OFFROUTE_HPA_DB")
+# HPA* cost-tile manifest (HPA-SPEC.md §8/§9, Phase H3 + H6a-pre). Empty manifest ->
+# HPA* never engages; Auto routing is byte-identical to the unified-graph path. Populated
+# from NAVI_OFFROUTE_HPA_DIR (manifest.json) or the legacy NAVI_OFFROUTE_HPA_DB (treated as
+# a single unbounded entry). See hpa_manifest.py.
+_HPA_MANIFEST = hpa_manifest.load()
 
 # Search radius for entry points (km)
 DEFAULT_SEARCH_RADIUS_KM = 50
@@ -916,15 +918,19 @@ class OffrouteRouter:
         origin_modes = np.array(sorted(MODE_INDEX[m] for m in start_eligible), dtype=np.int64)
         goal_modes = np.array(sorted(MODE_INDEX[m] for m in end_eligible), dtype=np.int64)
 
-        # HPA* fast path (Phase H3): when a tile DB is configured + covers the route, search
-        # the precomputed abstract chunk graph instead of flooding the full bbox. Whole-route
-        # fallback to the unified kernel below on any miss (HPA-SPEC.md §8/§9). When
-        # NAVI_OFFROUTE_HPA_DB is unset this block is skipped entirely (behaviour unchanged).
+        # HPA* fast path (Phase H3 + H6a-pre): consult the manifest for any tile DB(s) that
+        # cover this route's chunk range; with exactly one match, dispatch to the precomputed
+        # abstract graph instead of flooding the full bbox. Multi-region UNION across DBs is
+        # a future PR (needs astar_hpa_multimode signature change); for now multi-match logs
+        # the reason and falls through to the unified kernel. Whole-route fallback on any
+        # miss (HPA-SPEC.md §8/§9). With an empty manifest this block is skipped entirely.
         if self._hpa_eligible(boundary_mode, network_affinity):
+            db_paths = _HPA_MANIFEST.dbs_for_route_bbox(*meta["bounds"])
+            if len(db_paths) == 1:
                 _h0 = time.perf_counter()
                 _cache = {"raster": {}, "layer": {}}
                 hidx, hpath, hcost, hreason = astar_hpa_multimode(
-                HPA_TILE_DB, meta, start_lat, start_lon, end_lat, end_lon,
+                    db_paths[0], meta, start_lat, start_lon, end_lat, end_lon,
                     origin_modes, goal_modes, boundary_mode, network_affinity,
                     chunk_layer=lambda cx, cy, mi: self._hpa_chunk_layer(cx, cy, mi, _cache),
                     dem_reader=self.dem_reader)
@@ -933,7 +939,10 @@ class OffrouteRouter:
                                 len(_cache["raster"]), time.perf_counter() - _h0)
                     return self._render_unified_path(hpath, hcost, meta, boundary_mode)
                 logger.info("auto: HPA fallback reason=%s -> unified A*", hreason)
-        elif HPA_TILE_DB and os.path.exists(HPA_TILE_DB):
+            else:
+                _r = "multi_region" if len(db_paths) > 1 else "no_coverage"
+                logger.info("auto: HPA fallback reason=%s -> unified A*", _r)
+        elif _HPA_MANIFEST.enabled():
             _r = "boundary_mode" if boundary_mode != "pragmatic" else "affinity"
             logger.info("auto: HPA fallback reason=%s -> unified A*", _r)
 
@@ -1112,10 +1121,10 @@ class OffrouteRouter:
         }
 
     def _hpa_eligible(self, boundary_mode, network_affinity):
-        """HPA* engages only with a configured + existing tile DB, the default boundary mode,
+        """HPA* engages only with a non-empty tile-DB manifest, the default boundary mode,
         and no network_affinity — the tiles are pure-terrain/pragmatic (HPA-SPEC.md §8), so
         other configs would change the answer and must use the unified fallback."""
-        if not (HPA_TILE_DB and os.path.exists(HPA_TILE_DB)):
+        if not _HPA_MANIFEST.enabled():
             return False
         if boundary_mode != "pragmatic":
             return False

backend/services/navi_offroute/tests/test_hpa_manifest.py

@@ -0,0 +1,108 @@
+"""HPA tile-DB manifest pattern (hpa_manifest.py) unit tests."""
+import json
+import logging
+import os
+
+import pytest
+
+from services.navi_offroute import hpa_manifest as hm
+
+
+def _write(dirpath, payload):
+    with open(os.path.join(dirpath, "manifest.json"), "w") as f:
+        json.dump(payload, f)
+
+
+def _touch(dirpath, name):
+    p = os.path.join(dirpath, name)
+    open(p, "w").close()
+    return p
+
+
+def test_chunk_bounds_intersects():
+    cb = hm.ChunkBounds(min_x=-10, max_x=10, min_y=0, max_y=100)
+    assert cb.intersects(0, 0, 50, 50)            # point inside
+    assert cb.intersects(-20, -5, 50, 50)         # overlapping west edge
+    assert cb.intersects(-10, 10, 0, 100)         # exact match (inclusive)
+    assert not cb.intersects(11, 20, 50, 50)      # entirely east
+    assert not cb.intersects(0, 5, 101, 200)      # entirely north
+
+
+def test_load_from_dir_roundtrip_and_unbounded(tmp_path):
+    a = _touch(str(tmp_path), "a.db"); _touch(str(tmp_path), "world.db")
+    _write(str(tmp_path), {"version": 1, "tile_dbs": [
+        {"name": "a", "path": "a.db",
+         "chunk_bounds": {"min_x": -8550, "max_x": -8430, "min_y": 3100, "max_y": 3260}},
+        {"name": "world", "path": "world.db", "chunk_bounds": None},
+    ]})
+    m = hm._load_from_dir(str(tmp_path))
+    assert m.enabled() and len(m.entries) == 2
+    assert m.entries[0].abs_path == a
+    assert m.entries[0].chunk_bounds == hm.ChunkBounds(-8550, -8430, 3100, 3260)
+    assert m.entries[1].chunk_bounds is None
+
+
+def test_load_from_dir_fails_fast(tmp_path):
+    # missing manifest -> FileNotFoundError
+    with pytest.raises(FileNotFoundError, match="manifest not found"):
+        hm._load_from_dir(str(tmp_path))
+    # bad version -> ValueError
+    _write(str(tmp_path), {"version": 99, "tile_dbs": []})
+    with pytest.raises(ValueError, match="unsupported version"):
+        hm._load_from_dir(str(tmp_path))
+    # entry pointing at missing file -> FileNotFoundError
+    _write(str(tmp_path), {"version": 1, "tile_dbs": [
+        {"name": "ghost", "path": "ghost.db", "chunk_bounds": None}]})
+    with pytest.raises(FileNotFoundError, match="ghost.db"):
+        hm._load_from_dir(str(tmp_path))
+
+
+def test_dbs_for_chunks_lookup(tmp_path):
+    a = _touch(str(tmp_path), "a.db"); b = _touch(str(tmp_path), "b.db")
+    w = _touch(str(tmp_path), "world.db")
+    _write(str(tmp_path), {"version": 1, "tile_dbs": [
+        {"name": "a", "path": "a.db",
+         "chunk_bounds": {"min_x": 0, "max_x": 10, "min_y": 0, "max_y": 10}},
+        {"name": "b", "path": "b.db",
+         "chunk_bounds": {"min_x": 20, "max_x": 30, "min_y": 0, "max_y": 10}},
+        {"name": "world", "path": "world.db", "chunk_bounds": None},
+    ]})
+    m = hm._load_from_dir(str(tmp_path))
+    assert m.dbs_for_chunks(5, 5, 5, 5) == [a, w]          # hits a + unbounded
+    assert m.dbs_for_chunks(25, 25, 5, 5) == [b, w]        # hits b + unbounded
+    assert m.dbs_for_chunks(5, 25, 5, 5) == [a, b, w]      # straddles seam
+    assert m.dbs_for_chunks(100, 200, 0, 0) == [w]         # only unbounded
+
+
+def test_load_env_dir_wins_over_legacy(tmp_path, monkeypatch, caplog):
+    _touch(str(tmp_path), "world.db")
+    _write(str(tmp_path), {"version": 1, "tile_dbs": [
+        {"name": "w", "path": "world.db", "chunk_bounds": None}]})
+    legacy = _touch(str(tmp_path), "legacy.db")
+    monkeypatch.setenv(hm.ENV_DIR, str(tmp_path))
+    monkeypatch.setenv(hm.ENV_LEGACY, legacy)
+    with caplog.at_level(logging.WARNING, logger="services.navi_offroute.hpa_manifest"):
+        m = hm.load()
+    assert [e.name for e in m.entries] == ["w"]
+    assert "takes precedence" in caplog.text
+
+
+def test_load_env_legacy_synthesizes_unbounded(tmp_path, monkeypatch, caplog):
+    legacy = _touch(str(tmp_path), "legacy.db")
+    monkeypatch.delenv(hm.ENV_DIR, raising=False)
+    monkeypatch.setenv(hm.ENV_LEGACY, legacy)
+    with caplog.at_level(logging.WARNING, logger="services.navi_offroute.hpa_manifest"):
+        m = hm.load()
+    assert len(m.entries) == 1 and m.entries[0].chunk_bounds is None
+    assert m.entries[0].abs_path == legacy
+    assert "deprecated" in caplog.text
+
+
+def test_load_env_disabled_paths(tmp_path, monkeypatch):
+    # legacy points at missing file -> empty
+    monkeypatch.delenv(hm.ENV_DIR, raising=False)
+    monkeypatch.setenv(hm.ENV_LEGACY, str(tmp_path / "nope.db"))
+    assert not hm.load().enabled()
+    # neither var set -> empty
+    monkeypatch.delenv(hm.ENV_LEGACY, raising=False)
+    assert not hm.load().enabled()

backend/services/navi_offroute/tests/test_hpa_runtime.py

@@ -9,6 +9,7 @@ import pytest
 
 from services.navi_offroute import astar, hpa_build as hb
 import services.navi_offroute.router as router_mod
+from services.navi_offroute import hpa_manifest
 import services.navi_offroute.transitions as p_trans
 from services.navi_offroute.router import OffrouteRouter
 from services.navi_offroute.transitions import _latlon_to_pixel as _ll2px, _pixel_to_latlon as _px2ll
@@ -130,7 +131,8 @@ def _stub_router(monkeypatch):
 def test_hpa_dispatcher_uses_hpa_when_tile_db_set(tmp_path, monkeypatch, caplog):
     db = str(tmp_path / "tiles.db")
     _make_tile_db(db, _base_rows())                       # a real file so os.path.exists passes
-    monkeypatch.setattr(router_mod, "HPA_TILE_DB", db)
+    monkeypatch.setattr(router_mod, "_HPA_MANIFEST", hpa_manifest.HPAManifest(
+        entries=[hpa_manifest.TileDBEntry(name="test", abs_path=db, chunk_bounds=None)]))
 
     calls = []