feat(poller): plan-aware short-wait alerts

infra/lambda/poller/index.py

@@ -398,15 +398,56 @@ def filter_to_favoriters(
                 )
 
                 if historical_fire or forecast_fire:
+                    # Two alert sources can match the same user, same as
+                    # the DOWN path: favoriting the ride generally, OR
+                    # having it (still un-ridden) in TODAY's active plan.
+                    # The plan-aware framing is more actionable ("jump to
+                    # it now, it's cheaper than planned"), so it wins via
+                    # the resolver. Plan targets come from the active-plan
+                    # ride index, which only holds ride_sequence of ACTIVE
+                    # plans inside their window — dormant plans, completed
+                    # rides, and out-of-window hours never alert.
+                    low_wait_kwargs = {
+                        "ride_name": attr["name"],
+                        "park_name": attr["park_name"],
+                        "park_key": park_key,
+                        "wait_mins": new_wait,
+                        "typical_wait_mins": (
+                            typical_for_msg if historical_fire else None
+                        ),
+                        "forecast_wait_mins": (
+                            forecast_wait if forecast_fire else None
+                        ),
+                    }
                     favoriters = filter_to_favoriters(
                         subscribers, park_key, ride_id
                     )
-                    if favoriters:
+                    plan_targets = db.lookup_plan_targets(
+                        plan_ride_index, ride_id, attr["name"]
+                    )
+                    candidates: list[alert_routing.AlertCandidate] = []
+                    for target_user, target_plan in plan_targets:
+                        candidates.append(alert_routing.AlertCandidate(
+                            user_id=target_user,
+                            priority=alert_routing.PRIORITY_PLAN,
+                            notifier_fn=notifier.alert_plan_low_wait,
+                            kwargs={**low_wait_kwargs, "plan_id": target_plan},
+                        ))
+                    for fav_user in favoriters:
+                        candidates.append(alert_routing.AlertCandidate(
+                            user_id=fav_user,
+                            priority=alert_routing.PRIORITY_FAVORITE,
+                            notifier_fn=notifier.alert_low_wait,
+                            kwargs=dict(low_wait_kwargs),
+                        ))
+
+                    resolved = alert_routing.resolve_alert_recipients(candidates)
+                    if resolved:
                         # Cooldown is set per-ride (not per-recipient)
                         # — same pattern as DOWN. Set it only when we
                         # actually fan out so a window with no
-                        # favoriters can still alert the next user
-                        # who favorites the ride.
+                        # recipients can still alert the next user who
+                        # favorites/plans the ride.
                         db.mark_low_wait_alert_sent(ride_id)
                         # Log which baseline(s) tripped so post-hoc
                         # log analysis can audit false-positive rate
@@ -419,22 +460,20 @@ def filter_to_favoriters(
                         print(
                             f"[poller] {attr['name']} LOW WAIT "
                             f"({new_wait}m, {', '.join(triggers)}): "
-                            f"alerting {len(favoriters)} favoriters"
-                        )
-                        total_alerts += _fanout(
-                            favoriters, get_user_key,
-                            notifier.alert_low_wait,
-                            ride_name=attr["name"],
-                            park_name=attr["park_name"],
-                            park_key=park_key,
-                            wait_mins=new_wait,
-                            typical_wait_mins=(
-                                typical_for_msg if historical_fire else None
-                            ),
-                            forecast_wait_mins=(
-                                forecast_wait if forecast_fire else None
-                            ),
+                            f"{len(favoriters)} favoriters, "
+                            f"{len(plan_targets)} plan targets → "
+                            f"{len(resolved)} unique recipients"
                         )
+                        for target_user, candidate in resolved.items():
+                            user_key = get_user_key(target_user)
+                            if not user_key:
+                                print(
+                                    f"[poller] No pushover_user_key for "
+                                    f"user {target_user} — skipping"
+                                )
+                                continue
+                            if candidate.notifier_fn(user_key, **candidate.kwargs):
+                                total_alerts += 1
 
             # No status change → nothing more to do for this ride.
             if new_status == old_status:

infra/lambda/poller/notifier.py

@@ -161,6 +161,40 @@ def alert_low_wait(
     return _send(user_key, title, body, priority=0)
 
 
+def alert_plan_low_wait(
+    user_key: str,
+    ride_name: str,
+    park_name: str,
+    park_key: str,
+    wait_mins: int,
+    typical_wait_mins: int | None = None,
+    forecast_wait_mins: int | None = None,
+    plan_id: Optional[str] = None,
+) -> bool:
+    """Plan-aware sibling of alert_low_wait: the ride with the unusually
+    short wait is in the recipient's ACTIVE plan today (still in
+    ride_sequence — not yet ridden), so this is directly actionable:
+    jump to it now and ride it cheaper than planned.
+
+    Same signal + cooldown as alert_low_wait (one low-wait-class push
+    per ride per window); only the framing differs. Priority 0 —
+    an opportunity, not a disruption.
+    """
+    emoji = PARK_EMOJI.get(park_key, "🎢")
+    title = f"{emoji} Plan opportunity — {ride_name} low wait"
+    parts: list[str] = [
+        f"{ride_name} is at {wait_mins} min right now and it's in your "
+        f"plan today.",
+    ]
+    if typical_wait_mins is not None:
+        parts.append(f"Typical for this hour: ~{typical_wait_mins} min.")
+    if forecast_wait_mins is not None:
+        parts.append(f"Today's forecast: {forecast_wait_mins} min.")
+    parts.append("Good time to jump to it if you're close.")
+    body = f"{park_name}\n" + " ".join(parts)
+    return _send(user_key, title, body, priority=0)
+
+
 def alert_plan_disruption(
     user_key: str,
     ride_name: str,

infra/lambda/poller/tests/test_alert_routing.py

@@ -262,3 +262,44 @@ def test_down_path_candidate_construction_resolves_correctly(self):
 
         # Carol — plan only — gets the plan alert.
         assert resolved["carol"].kwargs["disruption_type"] == "went_down"
+
+
+class TestLowWaitScenario:
+    """Plan-aware LOW WAIT (added 2026-07-03) uses the same resolver as
+    DOWN/BACK UP: a user with the ride in today's active plan gets the
+    plan-framed opportunity alert; a favoriter-only user gets the generic
+    low-wait alert; a dual-source user gets the plan version."""
+
+    def test_plan_framing_beats_favorite_for_low_wait(self):
+        low_wait_kwargs = {
+            "ride_name": "Space Mountain",
+            "park_name": "Magic Kingdom",
+            "park_key": "magic_kingdom",
+            "wait_mins": 15,
+            "typical_wait_mins": 40,
+            "forecast_wait_mins": None,
+        }
+        candidates = [
+            AlertCandidate(
+                user_id="alice",  # dual-source: plan + favorite
+                priority=PRIORITY_PLAN,
+                notifier_fn=_fake_notifier,
+                kwargs={**low_wait_kwargs, "plan_id": "PLAN#p1"},
+            ),
+            AlertCandidate(
+                user_id="alice",
+                priority=PRIORITY_FAVORITE,
+                notifier_fn=_fake_notifier,
+                kwargs=dict(low_wait_kwargs),
+            ),
+            AlertCandidate(
+                user_id="bob",  # favorite only
+                priority=PRIORITY_FAVORITE,
+                notifier_fn=_fake_notifier,
+                kwargs=dict(low_wait_kwargs),
+            ),
+        ]
+        resolved = resolve_alert_recipients(candidates)
+        assert set(resolved.keys()) == {"alice", "bob"}
+        assert resolved["alice"].kwargs["plan_id"] == "PLAN#p1"  # plan wins
+        assert "plan_id" not in resolved["bob"].kwargs

infra/lambda/poller/tests/test_notifier.py

@@ -47,3 +47,27 @@ def fake_post(url, data=None, timeout=None):
 
     monkeypatch.setattr(notifier.requests, "post", fake_post)
     assert notifier._send("user-key-123", "title", "message") is False
+
+
+def test_alert_plan_low_wait_message(monkeypatch):
+    """Plan-aware low-wait: names the plan context, carries the wait +
+    baseline numbers, sends at priority 0 (opportunity, not disruption)."""
+    monkeypatch.setattr(notifier, "_get_app_token", lambda: "tok")
+    sent = {}
+
+    def fake_post(url, data=None, timeout=None):
+        sent.update(data)
+        return _FakeResp()
+
+    monkeypatch.setattr(notifier.requests, "post", fake_post)
+    ok = notifier.alert_plan_low_wait(
+        "user-key", ride_name="Space Mountain", park_name="Magic Kingdom",
+        park_key="magic_kingdom", wait_mins=15, typical_wait_mins=40,
+        plan_id="PLAN#p1",
+    )
+    assert ok is True
+    assert "Plan opportunity" in sent["title"]
+    assert "15 min" in sent["message"]
+    assert "in your plan today" in sent["message"]
+    assert "~40 min" in sent["message"]
+    assert sent["priority"] == 0