fix(roads511): distinguish full closures from partial restrictions

Adds explicit full-closure phrase matching (with partial-restriction
phrases as vetoes) for sub_type mapping in to_event(), so wording like
"All Shoulders Closed" or "Right Lane Closed" no longer gets broadcast
as a full road_closed event. Existing is_closure severity/summary logic
is untouched; the new is_full_closure_flag/_has_full_closure_language
helpers are additive and scoped to sub_type only.

work/meshai/env/roads511.py

@@ -21,6 +21,53 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+# Explicit full-road-closure language, matched case-insensitively. Used
+# ONLY by to_event()'s sub_type mapping (branch a) to decide "road_closed"
+# vs. "road_works" — distinct from (and stricter than) the adapter's
+# existing `is_closure` property, which loosely matches any "closed"
+# substring and therefore also flags partial-restriction wording (e.g.
+# "All Shoulders Closed") that the upstream owner explicitly does not want
+# broadcast as a closure.
+_FULL_CLOSURE_PHRASES = (
+    "all lanes closed",
+    "road closed",
+    "highway closed",
+    "fully closed",
+    "closed in both directions",
+)
+
+# Partial-restriction phrases that must NEVER be counted as a full closure.
+# Checked first so one of these vetoes a match even if closure wording
+# happens to appear elsewhere in the same description — e.g. "all
+# shoulders closed" must not fall through to a naive "closed" test, and
+# "one/right/left lane closed" must not be confused with "all lanes
+# closed" (note the singular "lane closed" vs. plural "lanes closed").
+_PARTIAL_RESTRICTION_PHRASES = (
+    "shoulder closed",
+    "shoulders closed",
+    "lane closed",
+    "ramp closed",
+    "alternating",
+    "lane restriction",
+)
+
+
+def _has_full_closure_language(description: str) -> bool:
+    """True if `description` contains explicit full-closure phrasing.
+
+    Case-insensitive. Partial-restriction phrases (shoulder/lane/ramp
+    closures, alternating traffic) are checked first and veto a match, so
+    they never register as a full closure regardless of other wording
+    present.
+    """
+    text = (description or "").lower()
+
+    if any(phrase in text for phrase in _PARTIAL_RESTRICTION_PHRASES):
+        return False
+
+    return any(phrase in text for phrase in _FULL_CLOSURE_PHRASES)
+
+
 class Roads511Adapter:
     """511 road conditions polling adapter."""
 
@@ -284,12 +331,21 @@ class Roads511Adapter:
                     lat = loc.get("latitude") or loc.get("lat")
                     lon = loc.get("longitude") or loc.get("lon") or loc.get("lng")
 
-            # Check closure status
-            is_closure = (
+            # Raw upstream full-closure signal, captured on its own (no
+            # description heuristic mixed in) so to_event() can build a
+            # stricter full-closure determination for sub_type mapping
+            # without touching is_closure below, which severity/summary
+            # logic elsewhere already depends on.
+            is_full_closure_flag = bool(
                 item.get("IsFullClosure") or
                 item.get("is_full_closure") or
                 item.get("fullClosure") or
-                item.get("closed") or
+                item.get("closed")
+            )
+
+            # Check closure status
+            is_closure = (
+                is_full_closure_flag or
                 "closure" in str(event_type).lower() or
                 "closed" in str(description).lower()
             )
@@ -323,6 +379,19 @@ class Roads511Adapter:
                 None
             )
 
+            # Granular ITD 511 v2 fields (roadwork vs. crash vs. closure
+            # discrimination). Defensive: absent on other states' 511 feeds.
+            event_sub_type = (
+                item.get("EventSubType") or
+                item.get("event_sub_type") or
+                ""
+            )
+            cause = (
+                item.get("Cause") or
+                item.get("cause") or
+                None
+            )
+
             # Default 6 hour TTL, refreshed every tick
             expires = now + 21600
 
@@ -346,7 +415,10 @@ class Roads511Adapter:
                 "properties": {
                     "roadway": roadway,
                     "is_closure": bool(is_closure),
+                    "is_full_closure_flag": is_full_closure_flag,
                     "last_updated": last_updated,
+                    "event_sub_type": event_sub_type,
+                    "cause": cause,
                 },
             }
 
@@ -411,10 +483,49 @@ class Roads511Adapter:
             # NOTE: Central-era rows used source 'itd_511' with 'idaho_511:event:*'
             # ids — a different keyspace the pre-seed intentionally does not cover.
             _external_id = evt.get("external_id") or event_id
+
+            # sub_type mapping: distinguish routine roadwork (suppressed by
+            # gating/incident.py's work-zone rule) from genuine news (crashes,
+            # closures, hazards). A construction-caused FULL closure is still
+            # real news and must broadcast, so a full closure wins regardless
+            # of EventType. Precedence:
+            #   1. full closure (raw flag OR explicit closure wording,
+            #      see _has_full_closure_language) -> road_closed (always)
+            #   2. EventType roadwork /
+            #      EventSubType has "construction" -> road_works (suppressed)
+            #   3. EventType accidentsAndIncidents -> incident
+            #   4. EventType closures              -> road_closed
+            #   5. else (specialEvents, unknown)   -> incident (fail open)
+            #
+            # NOTE: branch 1 deliberately does NOT reuse `_is_closure` above.
+            # `_is_closure` is a loose OR-chain (also used for severity and
+            # the summary text) that matches any "closed" substring in the
+            # description, including partial-restriction wording like "All
+            # Shoulders Closed" — routine construction the owner explicitly
+            # does not want broadcast. `_full_closure_for_mapping` is a
+            # stricter, mapping-only determination.
+            _event_type_val = str(evt.get("event_type") or "").strip().lower()
+            _event_sub_type_val = str(props.get("event_sub_type") or "").strip().lower()
+            _cause = props.get("cause")
+            _full_closure_for_mapping = bool(
+                props.get("is_full_closure_flag")
+            ) or _has_full_closure_language(_desc)
+
+            if _full_closure_for_mapping:
+                _sub_type = "road_closed"
+            elif _event_type_val == "roadwork" or "construction" in _event_sub_type_val:
+                _sub_type = "road_works"
+            elif _event_type_val == "accidentsandincidents":
+                _sub_type = "incident"
+            elif _event_type_val == "closures":
+                _sub_type = "road_closed"
+            else:
+                _sub_type = "incident"
+
             canonical_data = {
                 "external_id":    _external_id,   # 511_{itd_id}; enables durable pre-seed
                 "source":         "511",
-                "sub_type":       "road_closed" if _is_closure else "incident",
+                "sub_type":       _sub_type,
                 "road":           _roadway or None,
                 "direction":      None,   # not structured in native 511 feed
                 "from_loc":       None,
@@ -423,7 +534,7 @@ class Roads511Adapter:
                 "mile_end":       None,
                 "mile_marker":    None,
                 "lanes_affected": None,
-                "cause":          None,
+                "cause":          _cause,
                 "comment":        _desc[:200] if _desc else title,
                 "impact":         "all lanes closed" if _is_closure else None,
                 "county":         None,

work/tests/test_adapter_roads511.py

@@ -200,3 +200,331 @@ def test_to_event_missing_properties_returns_event(adapter):
 def test_to_event_does_not_raise_on_corrupted_dict(adapter):
     """Corrupted dict returns None without raising."""
     assert adapter.to_event({"garbage": True}) is None
+
+
+# ============================================================
+# EventType / EventSubType / Cause -> sub_type MAPPING TESTS
+#
+# Drives real upstream-shaped ITD 511 v2 payloads through the actual
+# _parse_event() -> to_event() pipeline (not the make_511_event() stored-
+# dict helper) so the EventSubType/Cause capture at parse time and the
+# sub_type precedence in to_event() are both exercised end to end.
+# ============================================================
+
+def _itd_raw(
+    event_id="RW-1001",
+    event_type="roadwork",
+    event_sub_type="roadConstruction",
+    cause="Construction",
+    description="Paving operations, expect delays",
+    roadway="I-84",
+    is_full_closure=False,
+    lat=43.5,
+    lon=-116.2,
+):
+    """Realistic raw ITD 511 v2 API item shape (PascalCase fields)."""
+    return {
+        "EventId": event_id,
+        "EventType": event_type,
+        "EventSubType": event_sub_type,
+        "Cause": cause,
+        "RoadwayName": roadway,
+        "Description": description,
+        "Latitude": lat,
+        "Longitude": lon,
+        "IsFullClosure": is_full_closure,
+        "LastUpdated": "2026-08-01T00:00:00Z",
+    }
+
+
+def _through_pipeline(adapter, raw_item):
+    """Parse a raw ITD item then translate to a pipeline Event, mirroring
+    the real adapter flow (_fetch -> _parse_event -> get_events -> to_event)."""
+    stored = adapter._parse_event(raw_item, time.time())
+    assert stored is not None
+    return adapter.to_event(stored)
+
+
+def test_mapping_roadwork_not_full_closure_is_road_works(adapter):
+    """Routine roadwork, not a full closure -> sub_type 'road_works' (suppressed)."""
+    raw = _itd_raw(event_type="roadwork", is_full_closure=False)
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+
+def test_mapping_roadwork_with_full_closure_is_road_closed(adapter):
+    """Construction-caused FULL closure is still genuine news -> 'road_closed',
+    and must NOT be suppressed like routine roadwork."""
+    raw = _itd_raw(
+        event_type="roadwork",
+        event_sub_type="bridgeConstruction",
+        description="Bridge replacement in progress",
+        is_full_closure=True,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_closed"
+
+
+def test_mapping_accidents_and_incidents_is_incident(adapter):
+    """EventType accidentsAndIncidents -> sub_type 'incident'."""
+    raw = _itd_raw(
+        event_id="INC-2001",
+        event_type="accidentsAndIncidents",
+        event_sub_type="crash",
+        cause="Collision",
+        description="Two vehicle collision blocking right lane",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "incident"
+
+
+def test_mapping_closures_is_road_closed(adapter):
+    """EventType closures -> sub_type 'road_closed'."""
+    raw = _itd_raw(
+        event_id="CL-3001",
+        event_type="closures",
+        event_sub_type="bridgeClosure",
+        cause="Maintenance",
+        description="Bridge out of service",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_closed"
+
+
+def test_mapping_special_events_is_incident(adapter):
+    """EventType specialEvents -> sub_type 'incident' (not suppressed)."""
+    raw = _itd_raw(
+        event_id="SE-4001",
+        event_type="specialEvents",
+        event_sub_type="parade",
+        cause=None,
+        description="Downtown parade route",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "incident"
+
+
+def test_mapping_unknown_or_missing_event_type_is_incident(adapter):
+    """Missing/unrecognized EventType fails open to 'incident' (not suppressed)."""
+    raw = {
+        "EventId": "UNK-5001",
+        "RoadwayName": "SH-21",
+        "Description": "Unusual event",
+        "Latitude": 43.5,
+        "Longitude": -115.5,
+    }
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "incident"
+
+
+def test_mapping_event_subtype_construction_overrides_non_roadwork_eventtype(adapter):
+    """EventSubType containing 'construction' forces 'road_works' even when
+    EventType itself is not literally 'roadwork'."""
+    raw = _itd_raw(
+        event_id="SE-6001",
+        event_type="specialEvents",
+        event_sub_type="bridgeConstruction",
+        cause="Construction",
+        description="Bridge deck construction nearby",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+
+def test_mapping_cause_is_carried_through_not_none(adapter):
+    """Cause from the upstream payload reaches canonical_data, not hardcoded None."""
+    raw = _itd_raw(cause="Weather")
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["cause"] == "Weather"
+    assert event.data["cause"] is not None
+
+
+def test_mapping_missing_cause_is_none(adapter):
+    """No Cause field upstream -> canonical cause is None, not a crash."""
+    raw = _itd_raw(cause=None)
+    del raw["Cause"]
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["cause"] is None
+
+
+# ============================================================
+# END-TO-END GATING TEST — proves a real roadwork event, run through the
+# actual adapter pipeline, is suppressed by gating/incident.py's work-zone
+# rule, while a real crash event from the same pipeline still broadcasts.
+# ============================================================
+
+def test_roadwork_event_suppressed_by_incident_gate_end_to_end(adapter):
+    """A real ITD roadwork payload, parsed and translated by the adapter,
+    must be suppressed at the gating layer (not just mapped to the right
+    sub_type in isolation)."""
+    from meshai.notifications.gating.incident import decide
+
+    raw = _itd_raw(event_id="RW-7001", event_type="roadwork", is_full_closure=False)
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+    result = decide(dict(event.data), source="511", now=time.time())
+    assert result.broadcast is False
+    assert result.lifecycle == "suppress"
+
+
+def test_crash_event_still_broadcasts_through_incident_gate_end_to_end(adapter):
+    """A real ITD crash payload, parsed and translated by the adapter,
+    still broadcasts (gating layer is unaffected for non-work-zone sub_types)."""
+    from meshai.notifications.gating.incident import decide
+
+    raw = _itd_raw(
+        event_id="INC-7002",
+        event_type="accidentsAndIncidents",
+        event_sub_type="crash",
+        description="Vehicle collision, right lane blocked",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "incident"
+
+
+# ============================================================
+# FULL-CLOSURE vs. PARTIAL-RESTRICTION LANGUAGE TESTS
+#
+# The loose `is_closure` property (also used for severity/summary) matches
+# any "closed" substring, including partial-restriction wording like "All
+# Shoulders Closed" that ITD explicitly does not want broadcast. The
+# sub_type mapping's full-closure branch must use a stricter,
+# mapping-only determination instead.
+# ============================================================
+
+def test_all_shoulders_closed_roadwork_is_road_works(adapter):
+    """'All Shoulders Closed' is a partial restriction, not a full closure
+    -> stays 'road_works' (suppressed), despite containing 'closed'."""
+    raw = _itd_raw(
+        event_id="RW-8001",
+        event_type="roadwork",
+        description="All Shoulders Closed",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+
+@pytest.mark.parametrize(
+    "description",
+    ["One lane closed", "Right lane closed", "Left lane closed"],
+)
+def test_single_lane_closed_roadwork_is_road_works(adapter, description):
+    """Single-lane restriction wording -> stays 'road_works' (suppressed)."""
+    raw = _itd_raw(
+        event_id="RW-8002",
+        event_type="roadwork",
+        description=description,
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+
+def test_ramp_closed_roadwork_is_road_works(adapter):
+    """'Ramp closed' is a partial restriction -> stays 'road_works' (suppressed)."""
+    raw = _itd_raw(
+        event_id="RW-8003",
+        event_type="roadwork",
+        description="Ramp closed for repaving",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+
+def test_all_lanes_closed_roadwork_is_road_closed(adapter):
+    """'All lanes closed' is a genuine full closure -> 'road_closed', and
+    must STILL BROADCAST (this is the safety case: a real closure phrased
+    as roadwork must not be suppressed by the work-zone gate)."""
+    from meshai.notifications.gating.incident import decide
+
+    raw = _itd_raw(
+        event_id="RW-8004",
+        event_type="roadwork",
+        description="All lanes closed for bridge demolition",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_closed"
+
+    result = decide(dict(event.data), source="511", now=time.time())
+    assert result.broadcast is True
+    assert result.lifecycle != "suppress"
+
+
+def test_is_full_closure_flag_true_no_closure_words_is_road_closed(adapter):
+    """IsFullClosure=true + roadwork, with a description containing no
+    closure wording at all -> still 'road_closed' via the raw flag."""
+    raw = _itd_raw(
+        event_id="RW-8005",
+        event_type="roadwork",
+        description="Bridge deck replacement in progress",
+        is_full_closure=True,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_closed"
+
+
+def test_road_closed_due_to_rock_slide_incident_is_road_closed(adapter):
+    """Explicit full-closure wording on an accidentsAndIncidents event
+    still wins the mapping -> 'road_closed'."""
+    raw = _itd_raw(
+        event_id="INC-8006",
+        event_type="accidentsAndIncidents",
+        event_sub_type="hazard",
+        cause="Rock slide",
+        description="Road closed due to rock slide",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_closed"
+
+
+def test_lane_restrictions_roadwork_is_road_works(adapter):
+    """'Lane restrictions' wording -> stays 'road_works' (suppressed)."""
+    raw = _itd_raw(
+        event_id="RW-8007",
+        event_type="roadwork",
+        description="Lane restrictions in effect through Friday",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"
+
+
+def test_alternating_traffic_roadwork_is_road_works(adapter):
+    """'Alternating' one-lane traffic control -> stays 'road_works' (suppressed)."""
+    raw = _itd_raw(
+        event_id="RW-8008",
+        event_type="roadwork",
+        description="Alternating traffic controlled by flaggers",
+        is_full_closure=False,
+    )
+    event = _through_pipeline(adapter, raw)
+    assert event is not None
+    assert event.data["sub_type"] == "road_works"