feat(wb-alerts): Phase 1 — per-channel dynamic trust for checkpoint reports

Checkpoint source_trust was a static {admin:0.9, crowd:0.4} keyed only on
last_source_type — every crowd report scored 0.4 regardless of which channel
reported it, so the trust number carried no signal (audit leak #1). But the
checkpoint channels (ahwalaltreq 0.75, a7walstreet 0.75, road_jehad 0.70, ...)
already have learner-tuned reliability weights in channel_reliability.

Thread that weight into crowd trust: provenance.source_trust(last_source_type,
channel_reliability) — admin stays authoritative (0.9); crowd is scored by the
channel's reliability bounded [0.2, 0.85] (never reaches admin, keeps a floor);
falls back to the flat 0.4 prior when no weight is known. provenance stays
I/O-free; routers/v2 resolves the weights once per request via the new
get_channel_reliability_map() (degrades to {} if the table is absent).

Tests: +9 (provenance + v2 wiring); suite 147 passed, FP audit 96/96.

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

Author: 2 people

services/westbank-alerts/app/database.py

@@ -732,6 +732,21 @@ async def get_channel_reliability(channel: Optional[str] = None):
     return out
 
 
+async def get_channel_reliability_map() -> dict:
+    """All channel reliability weights as {lower(channel): weight} — one query,
+    for batch construction of checkpoint envelopes (Phase 1 dynamic trust).
+
+    Reliability is enrichment: if the table is unavailable, return {} so callers
+    fall back to the flat crowd prior rather than failing the checkpoint feed.
+    """
+    try:
+        async with get_alerts_db() as db:
+            cur = await db.execute("SELECT lower(channel), weight FROM channel_reliability")
+            return {row[0]: float(row[1]) for row in await cur.fetchall()}
+    except aiosqlite.OperationalError:
+        return {}
+
+
 # ── Security vocab candidates ─────────────────────────────────────────────────
 
 async def insert_security_vocab_candidate(

services/westbank-alerts/app/routers/v2.py

@@ -14,6 +14,7 @@
 from .. import checkpoint_db as cpdb
 from .. import incident_db
 from ..config import settings
+from ..database import get_channel_reliability_map
 from ..db_pool import get_alerts_db
 from ..serving import provenance as P
 from ..serving import gateways as GW
@@ -23,11 +24,16 @@
 _STALE = settings.CHECKPOINT_STALE_HOURS
 
 
-def _cp_envelope(cp: dict) -> dict:
+def _cp_envelope(cp: dict, rel_map: Optional[dict] = None) -> dict:
     # Use the RAW last_updated (None when never reported) so freshness can return
     # band "none" instead of treating a never-reported checkpoint as fresh.
     src = {**cp, "last_updated": cp.get("last_updated_iso")}
-    return P.checkpoint_envelope(src, stale_hours=_STALE)
+    # Phase 1: a crowd report's trust reflects the reporting channel's observed
+    # reliability (resolved here, not inside the I/O-free provenance layer).
+    rel = None
+    if rel_map:
+        rel = rel_map.get((cp.get("source_channel") or "").lower())
+    return P.checkpoint_envelope(src, stale_hours=_STALE, channel_reliability=rel)
 
 
 @router.get("/checkpoints")
@@ -36,7 +42,8 @@ async def v2_checkpoints(
     status: Optional[str] = Query(None, description="filter by effective_status"),
 ):
     cps = await cpdb.get_all_checkpoints(region=region)
-    envs = [_cp_envelope(c) for c in cps]
+    rel_map = await get_channel_reliability_map()
+    envs = [_cp_envelope(c, rel_map) for c in cps]
     if status:
         envs = [e for e in envs if e["effective_status"] == status]
     return {"checkpoints": envs, "total": len(envs),
@@ -46,9 +53,10 @@ async def v2_checkpoints(
 @router.get("/checkpoints/geojson")
 async def v2_checkpoints_geojson(region: Optional[str] = Query(None)):
     cps = await cpdb.get_all_checkpoints(region=region)
+    rel_map = await get_channel_reliability_map()
     features = []
     for c in cps:
-        env = _cp_envelope(c)
+        env = _cp_envelope(c, rel_map)
         lat, lon = env["coordinates"]["lat"], env["coordinates"]["lon"]
         if lat is None or lon is None:
             continue

services/westbank-alerts/app/serving/provenance.py

@@ -14,6 +14,12 @@
 LIVE_HOURS = 1.0  # <= this age → "live"
 
 _TRUST = {"admin": 0.9, "crowd": 0.4}
+# Phase 1: a crowd report's trust is driven by the reporting channel's observed
+# reliability (resolved by the caller), bounded so it keeps a floor of signal
+# and never reaches admin (0.9). Falls back to the flat 0.4 prior when the
+# caller has no reliability for the channel.
+_CROWD_TRUST_FLOOR = 0.2
+_CROWD_TRUST_CEIL = 0.85
 
 
 def _as_datetime(value: Union[datetime, str, None]) -> Optional[datetime]:
@@ -82,7 +88,24 @@ def effective_status(
     return cp.get("status") or "unknown"
 
 
-def source_trust(last_source_type: Optional[str]) -> float:
+def source_trust(
+    last_source_type: Optional[str],
+    channel_reliability: Optional[float] = None,
+) -> float:
+    """Trust score for a checkpoint's most recent report.
+
+    admin reports stay authoritative (0.9). A crowd report is scored by the
+    reporting channel's reliability weight when the caller supplies one
+    (bounded to [_CROWD_TRUST_FLOOR, _CROWD_TRUST_CEIL] so it carries a floor
+    of signal and never reaches admin); otherwise the flat 0.4 prior is used.
+    """
+    if last_source_type == "admin":
+        return _TRUST["admin"]
+    if last_source_type == "crowd":
+        if channel_reliability is None:
+            return _TRUST["crowd"]
+        return round(min(_CROWD_TRUST_CEIL,
+                         max(_CROWD_TRUST_FLOOR, channel_reliability)), 3)
     return _TRUST.get(last_source_type or "", 0.0)
 
 
@@ -91,6 +114,7 @@ def checkpoint_envelope(
     *,
     stale_hours: float = DEFAULT_STALE_HOURS,
     now: Optional[datetime] = None,
+    channel_reliability: Optional[float] = None,
 ) -> dict:
     """Canonical checkpoint record for the /v2 feeds and route endpoints.
 
@@ -125,7 +149,7 @@ def checkpoint_envelope(
         "freshness": fresh,
         "source_trust": {
             "last_source_type": cp.get("last_source_type"),
-            "trust": source_trust(cp.get("last_source_type")),
+            "trust": source_trust(cp.get("last_source_type"), channel_reliability),
         },
         "provenance": {
             "last_msg_id": cp.get("last_msg_id"),

services/westbank-alerts/test_provenance.py

@@ -77,6 +77,27 @@ def test_trust_admin_beats_crowd_beats_none():
     assert P.source_trust("admin") > P.source_trust("crowd") > P.source_trust(None)
 
 
+def test_trust_crowd_uses_channel_reliability_when_provided():
+    # Phase 1: a crowd report from a reputable checkpoint channel beats the
+    # flat 0.4 prior — trust must discriminate by channel.
+    assert P.source_trust("crowd", channel_reliability=0.75) == 0.75
+
+
+def test_trust_crowd_defaults_to_flat_prior_without_reliability():
+    # Backward compatible: when no reliability is resolved, keep the 0.4 prior.
+    assert P.source_trust("crowd") == 0.4
+
+
+def test_trust_crowd_capped_below_admin_and_floored():
+    # Crowd never reaches admin (0.9); a near-zero channel keeps a little signal.
+    assert P.source_trust("crowd", channel_reliability=0.99) == 0.85
+    assert P.source_trust("crowd", channel_reliability=0.05) == 0.2
+
+
+def test_trust_admin_stays_authoritative_regardless_of_channel():
+    assert P.source_trust("admin", channel_reliability=0.6) == 0.9
+
+
 # ── checkpoint_envelope ──────────────────────────────────────────────────────
 
 def _cp(**over):
@@ -113,6 +134,14 @@ def test_checkpoint_envelope_stale_row_effective_unknown():
     assert env["effective_status"] == "unknown"
 
 
+def test_checkpoint_envelope_threads_channel_reliability():
+    # Phase 1: a crowd report's trust reflects the channel's reliability weight,
+    # resolved by the caller and passed in (provenance stays I/O-free).
+    env = P.checkpoint_envelope(_cp(last_source_type="crowd"),
+                                channel_reliability=0.75, now=NOW)
+    assert env["source_trust"]["trust"] == 0.75
+
+
 def test_checkpoint_envelope_permanent_closure_marks_closed_even_if_stale():
     env = P.checkpoint_envelope(
         _cp(status="open", permanent_status="closed_since:2023-10",

services/westbank-alerts/test_v2_trust.py

@@ -0,0 +1,37 @@
+"""Phase 1: the /v2 checkpoint envelope threads per-channel reliability into
+the crowd trust score. _cp_envelope is pure given (cp, rel_map) — no DB — so
+the rel_map lookup, case-insensitive channel match, and fallback are unit-tested
+here; the bulk DB getter (get_channel_reliability_map) is exercised live.
+"""
+import os
+os.environ.setdefault("API_SECRET_KEY", "test-secret-key-0123456789abcdef")
+
+from app.routers import v2
+
+_REL = {"ahwalaltreq": 0.75, "rt_arabic": 0.45}
+
+
+def _cp(**over):
+    base = {
+        "canonical_key": "x", "source_channel": "AhwalAlTreq",
+        "last_source_type": "crowd", "status": "open", "last_updated_iso": None,
+    }
+    base.update(over)
+    return base
+
+
+def test_crowd_trust_uses_channel_weight_case_insensitive():
+    # "AhwalAlTreq" matches the lowercased "ahwalaltreq" key.
+    assert v2._cp_envelope(_cp(), _REL)["source_trust"]["trust"] == 0.75
+
+
+def test_crowd_trust_falls_back_for_unknown_channel():
+    assert v2._cp_envelope(_cp(source_channel="randomchan"), _REL)["source_trust"]["trust"] == 0.4
+
+
+def test_admin_trust_unaffected_by_channel_weight():
+    assert v2._cp_envelope(_cp(last_source_type="admin"), _REL)["source_trust"]["trust"] == 0.9
+
+
+def test_no_rel_map_preserves_legacy_crowd_prior():
+    assert v2._cp_envelope(_cp())["source_trust"]["trust"] == 0.4