fix(compare): wire compare intent into SSE handler

The compare intent was classified by the planner but silently fell
through to single_address routing in api_agent_stream because (a)
"compare" was not in INTENTS so _validate() defaulted to
single_address, and (b) there was no routing case in the SSE handler.

Changes:
- app/planner.py: add "compare" to INTENTS, SPECIALISTS (compare is
applicable to all single_address specialists), PLAN_SCHEMA_DESC
hard-rules, _required_specialists, _default_specialists, and
_validate fallback. Planner now emits intent=compare with two
address targets for "compare X vs Y" queries.
- web/main.py: add _run_compare() helper that runs the full
single_address specialist suite sequentially for each target via
i_addr.run() and merges the two paragraphs into one Markdown
document (## PLACE A / ## PLACE B sections). Add compare routing
case to both api_agent_stream and api_agent endpoints.

Verified: "Compare 80 Pioneer Street, Brooklyn to 100 Gold Street,
Manhattan" produces a two-target briefing (2820 chars) with both
addresses cited and Mellea grounding passing. 38 step events (19
per address), no errors.

Map limitation: map re-centers to the last-geocoded address (PLACE B).
Both places appear in the briefing text. Dual-marker map requires
a RipMap prop change (deferred).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/planner.py

@@ -65,24 +65,32 @@ INTENTS = {
         "DOB construction permits inside it, cross-reference each project "
         "with Sandy + DEP flood layers, return a flagged-projects list."
     ),
+    "compare": (
+        "Use ONLY when the query explicitly compares TWO specific street "
+        "ADDRESSES (e.g. 'compare 80 Pioneer St Brooklyn to 100 Gold St "
+        "Manhattan', 'which is riskier: X or Y?', 'X vs Y flood risk'). "
+        "Extract BOTH full street addresses into targets as two separate "
+        "{type: 'address', text: ...} objects. Run the full single-address "
+        "specialist suite for each."
+    ),
 }
 
 SPECIALISTS = {
     # name: (description, which intents may invoke it)
-    "geocode":       ("Resolve address text to lat/lon via NYC DCP Geosearch.",     ["single_address"]),
+    "geocode":       ("Resolve address text to lat/lon via NYC DCP Geosearch.",     ["single_address", "compare"]),
     "nta_resolve":   ("Resolve a neighborhood or borough name to NTA polygon(s).",  ["neighborhood"]),
-    "sandy":         ("2012 Sandy inundation extent (point-in-polygon or % of NTA).", ["single_address", "neighborhood"]),
-    "dep_stormwater":("DEP Stormwater Maps — 3 modeled scenarios.",                ["single_address", "neighborhood"]),
-    "floodnet":      ("Live FloodNet ultrasonic sensors + trigger history.",      ["single_address", "neighborhood", "live_now"]),
-    "nyc311":        ("NYC 311 flood-related complaints in buffer or polygon.",    ["single_address", "neighborhood"]),
-    "noaa_tides":    ("Live NOAA Battery / Kings Pt / Sandy Hook water level.",   ["single_address", "neighborhood", "live_now"]),
-    "nws_alerts":    ("Live NWS active flood-relevant alerts at point.",           ["single_address", "neighborhood", "live_now"]),
-    "nws_obs":       ("Live NWS hourly precip from nearest ASOS station.",         ["single_address", "neighborhood", "live_now"]),
-    "ttm_forecast":  ("Granite TTM r2 surge-residual nowcast at the Battery.",     ["single_address", "neighborhood", "live_now"]),
-    "microtopo":     ("LiDAR-derived terrain (HAND, TWI, percentile) at point or aggregated over polygon.", ["single_address", "neighborhood"]),
-    "ida_hwm":       ("USGS Hurricane Ida 2021 high-water marks proximity.",       ["single_address", "neighborhood"]),
-    "prithvi":       ("Prithvi-EO 2.0 Hurricane Ida 2021 satellite flood polygons.", ["single_address", "neighborhood"]),
-    "rag":           ("Retrieve relevant agency-report passages over the policy corpus.", ["single_address", "neighborhood", "development_check"]),
+    "sandy":         ("2012 Sandy inundation extent (point-in-polygon or % of NTA).", ["single_address", "neighborhood", "compare"]),
+    "dep_stormwater":("DEP Stormwater Maps — 3 modeled scenarios.",                ["single_address", "neighborhood", "compare"]),
+    "floodnet":      ("Live FloodNet ultrasonic sensors + trigger history.",      ["single_address", "neighborhood", "live_now", "compare"]),
+    "nyc311":        ("NYC 311 flood-related complaints in buffer or polygon.",    ["single_address", "neighborhood", "compare"]),
+    "noaa_tides":    ("Live NOAA Battery / Kings Pt / Sandy Hook water level.",   ["single_address", "neighborhood", "live_now", "compare"]),
+    "nws_alerts":    ("Live NWS active flood-relevant alerts at point.",           ["single_address", "neighborhood", "live_now", "compare"]),
+    "nws_obs":       ("Live NWS hourly precip from nearest ASOS station.",         ["single_address", "neighborhood", "live_now", "compare"]),
+    "ttm_forecast":  ("Granite TTM r2 surge-residual nowcast at the Battery.",     ["single_address", "neighborhood", "live_now", "compare"]),
+    "microtopo":     ("LiDAR-derived terrain (HAND, TWI, percentile) at point or aggregated over polygon.", ["single_address", "neighborhood", "compare"]),
+    "ida_hwm":       ("USGS Hurricane Ida 2021 high-water marks proximity.",       ["single_address", "neighborhood", "compare"]),
+    "prithvi":       ("Prithvi-EO 2.0 Hurricane Ida 2021 satellite flood polygons.", ["single_address", "neighborhood", "compare"]),
+    "rag":           ("Retrieve relevant agency-report passages over the policy corpus.", ["single_address", "neighborhood", "development_check", "compare"]),
     "dob_permits":   ("Active NYC DOB construction permits inside a polygon, each cross-referenced with Sandy + DEP flood scenarios. Use for 'what are they building' / 'projects in progress' queries.", ["development_check"]),
 }
 
@@ -117,6 +125,7 @@ Hard rules:
 - For intent=neighborhood: ALWAYS include "nta_resolve". Skip "geocode". Include polygon-capable specialists.
 - For intent=live_now: ONLY live specialists. Skip historic/modeled (sandy, dep_*, ida_hwm, prithvi).
 - For intent=development_check: ALWAYS include "nta_resolve" AND "dob_permits". Sandy + DEP are also useful so the model can compare project locations to flood layers.
+- For intent=compare: ALWAYS include "geocode". Extract BOTH street addresses into targets — the executor runs the full specialist suite once per address. Targets must be exactly 2 items, both type="address".
 - IMPORTANT — TARGETS: extract neighborhood/borough names directly from the query text. If the query says "in Gowanus", "what about Brighton Beach", "around Carroll Gardens", etc., the target MUST be {"type": "nta", "text": "<the place name>"}. Use {"type": "nyc"} ONLY when the query mentions NYC as a whole and no specific place. Failing to extract a place name will cause the executor to give up — be explicit.
 - "targets" is a list because the user may name multiple places (e.g. "compare Brighton Beach and Coney Island").
 - "rationale" is one short sentence — what your reasoning was.
@@ -259,6 +268,13 @@ def _validate(d: dict[str, Any], raw_query: str) -> Plan:
             targets = [{"type": "address", "text": raw_query}]
         elif intent == "neighborhood":
             targets = [{"type": "nta", "text": raw_query}]
+        elif intent == "compare":
+            # Planner failed to extract two addresses — treat whole query as
+            # single address so the caller gets at least one result rather
+            # than a confusing empty response.
+            log.warning("compare intent but no valid targets extracted; "
+                        "falling back to single raw query")
+            targets = [{"type": "address", "text": raw_query}]
         else:
             targets = [{"type": "nyc", "text": "NYC"}]
 
@@ -295,11 +311,13 @@ def _required_specialists(intent: str) -> list[str]:
         return ["nws_alerts", "noaa_tides"]
     if intent == "development_check":
         return ["nta_resolve", "dob_permits", "sandy", "dep_stormwater"]
+    if intent == "compare":
+        return ["geocode", "sandy", "dep_stormwater", "microtopo"]
     return []
 
 
 def _default_specialists(intent: str) -> list[str]:
-    if intent == "single_address":
+    if intent in ("single_address", "compare"):
         return ["geocode", "sandy", "dep_stormwater", "floodnet", "nyc311",
                 "noaa_tides", "nws_alerts", "nws_obs", "ttm_forecast",
                 "microtopo", "ida_hwm", "prithvi", "rag"]

web/main.py

@@ -497,6 +497,85 @@ async def stream(q: str, request: Request):
                                       "X-Accel-Buffering": "no"})
 
 
+def _run_compare(p, raw_query: str, out_q, i_addr) -> dict:
+    """Run the compare intent: execute the full single_address specialist
+    suite sequentially for each target, then merge the two paragraphs into
+    one Markdown document clearly labelled PLACE A and PLACE B.
+
+    Sequential execution is required because the FSM uses thread-local hooks
+    (set_strict_mode, set_token_callback) — concurrent runs on the same
+    thread would corrupt the hooks. See app/intents/single_address.py.
+
+    Step events from each target are forwarded to out_q tagged with a
+    `target_label` key so the trace UI can optionally group them, but the
+    existing trace UI ignores unknown keys gracefully."""
+    from app.planner import Plan
+
+    addr_targets = [t for t in p.targets if t.get("type") == "address"]
+    if len(addr_targets) < 2:
+        # Fallback: only one (or zero) address extracted — run as single_address
+        return i_addr.run(p, raw_query, progress_q=out_q, strict=True)
+
+    results = []
+    for idx, target in enumerate(addr_targets[:2]):
+        label = "PLACE A" if idx == 0 else "PLACE B"
+        addr_text = target["text"]
+        # Synthetic single-address plan for this target
+        sub_plan = Plan(
+            intent="single_address",
+            targets=[{"type": "address", "text": addr_text}],
+            specialists=p.specialists,
+            rationale=p.rationale,
+        )
+
+        if out_q is not None:
+            # Wrap out_q to tag step events with the target label so the
+            # trace UI can optionally group them; token/mellea_attempt pass
+            # through untagged so the SvelteKit briefing buffer works.
+            _label = label
+            _q = out_q
+            class _TaggedQ:
+                def put(self, ev):
+                    if ev.get("kind") == "step":
+                        _q.put({**ev, "target_label": _label})
+                    else:
+                        _q.put(ev)
+            effective_q = _TaggedQ()
+        else:
+            effective_q = None
+
+        result = i_addr.run(sub_plan, addr_text, progress_q=effective_q, strict=True)
+        results.append((label, addr_text, result))
+
+    # Merge: produce one paragraph with both place sections.
+    parts = []
+    for label, addr_text, res in results:
+        para = (res.get("paragraph") or "").strip()
+        parts.append(f"## {label}: {addr_text}\n\n{para}")
+    merged_paragraph = "\n\n---\n\n".join(parts)
+
+    # Combine Mellea metadata: sum attempts, union passed/failed.
+    def _merge_mellea(a, b):
+        def _lst(m, k): return m.get(k) or []
+        return {
+            "rerolls": (a.get("rerolls") or 0) + (b.get("rerolls") or 0),
+            "n_attempts": (a.get("n_attempts") or 0) + (b.get("n_attempts") or 0),
+            "requirements_passed": list(set(_lst(a, "requirements_passed") + _lst(b, "requirements_passed"))),
+            "requirements_failed": list(set(_lst(a, "requirements_failed") + _lst(b, "requirements_failed"))),
+            "requirements_total": max(a.get("requirements_total") or 0, b.get("requirements_total") or 0),
+        }
+
+    mellea_a = results[0][2].get("mellea") or {}
+    mellea_b = results[1][2].get("mellea") or {}
+    return {
+        "paragraph": merged_paragraph,
+        "mellea": _merge_mellea(mellea_a, mellea_b),
+        "intent": "compare",
+        "targets": [{"label": lbl, "address": addr} for lbl, addr, _ in results],
+        "tier": results[0][2].get("tier"),
+    }
+
+
 @app.get("/api/agent")
 def api_agent(q: str):
     """Agentic endpoint: take a natural-language query, plan it via
@@ -523,7 +602,9 @@ def api_agent(q: str):
                        "requirements_total": 0},
             "status": "not_implemented",
         })
-    if p.intent == "development_check":
+    if p.intent == "compare":
+        out = _run_compare(p, q, None, i_addr)
+    elif p.intent == "development_check":
         out = i_dev.run(p, q, strict=True)
     elif p.intent == "neighborhood":
         out = i_nbhd.run(p, q, strict=True)
@@ -568,6 +649,8 @@ async def api_agent_stream(q: str):
                                "requirements_total": 0},
                     "status": "not_implemented",
                 }
+            elif p.intent == "compare":
+                final = _run_compare(p, q, out_q, i_addr)
             elif p.intent == "development_check":
                 final = i_dev.run(p, q, progress_q=out_q, strict=True)
             elif p.intent == "neighborhood":