fix: cold-start, keepalives, vLLM readiness probe, fast Ollama fallback timeout

app/fsm.py

@@ -13,18 +13,20 @@ import time
 from typing import Any
 
 import geopandas as gpd
-from burr.core import ApplicationBuilder, State, action
+from burr.core import ApplicationBuilder, State, action, expr
+from burr.lifecycle import PostRunStepHook
+from burr.tracking import LocalTrackingClient
 from shapely.geometry import Point
 
 from app import emissions
-from app.context import floodnet, microtopo, noaa_tides, nws_alerts, nws_obs, nyc311
+from app.context import floodnet, microtopo, noaa_tides, npcc4_slr, nws_alerts, nws_obs, nyc311
 from app.energy import estimate as energy_estimate
 from app.flood_layers import dep_stormwater, ida_hwm, prithvi_water, sandy_inundation
 from app.geocode import geocode_one
 from app.live import floodnet_forecast as fn_forecast
 from app.live import ttm_forecast
 from app.rag import retrieve as rag_retrieve
-from app.reconcile import reconcile as run_reconcile
+from app.reconcile import citations_from_docs, reconcile as run_reconcile
 from app.registers import doe_schools as r_schools
 from app.registers import doh_hospitals as r_hospitals
 from app.registers import mta_entrances as r_mta
@@ -119,14 +121,24 @@ def _current_planner_intent() -> str | None:
     return getattr(_FSM_LOCAL, "planner_intent", None)
 
 
-# Canonical Burr: one action per specialist, sequential transitions.
-# A previous version of this module wrapped 16 specialists in a single
-# fan-out action that ran them concurrently in a ThreadPoolExecutor;
-# that path was removed because it sometimes hung after the fan-out
-# completed (Burr-internal post-action cleanup with custom executors)
-# and made the trace UI's per-step timing harder to reason about.
-# Parallelism, when wanted, belongs at the inference layer
-# (vLLM / Ollama NUM_PARALLEL), not the FSM.
+class StepEventHook(PostRunStepHook):
+    """Burr lifecycle hook — fires after each action and pushes a
+    ``("step", rec)`` tuple onto a caller-supplied queue.
+
+    Replaces the manual ``seen_keys`` deduplication loop in ``iter_steps``.
+    Pass ``queue=None`` to construct a no-op hook (non-streaming paths)."""
+
+    def __init__(self, queue=None):
+        self._q = queue
+
+    def post_run_step(self, *, state: State, action, result, exception, **_kw):
+        if self._q is None:
+            return
+        trace = state.get("trace") or []
+        if not trace:
+            return
+        self._q.put(("step", trace[-1]))
+
 
 def _step(state: State, name: str) -> dict[str, Any]:
     """Append a step record to the trace; returns the dict so the action
@@ -137,6 +149,229 @@ def _step(state: State, name: str) -> dict[str, Any]:
     return rec, trace
 
 
+def _make_rec(name: str) -> dict[str, Any]:
+    """Trace record for use outside of Burr state (parallel workers)."""
+    return {"step": name, "started_at": time.time(), "ok": None}
+
+
+# ---------------------------------------------------------------------------
+# Cornerstone parallel helpers — plain functions, no State dependency.
+# Each returns (state_key, value, trace_rec). step_cornerstone fans them
+# out via ThreadPoolExecutor and merges results into Burr state in one shot.
+# Using a single Burr action with internal threads avoids the previous hang
+# (which was caused by Burr-internal post-action cleanup racing with a
+# custom executor passed to ApplicationBuilder).
+# ---------------------------------------------------------------------------
+
+def _run_sandy(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("sandy_inundation")
+    try:
+        if not _in_nyc(lat, lon):
+            rec["ok"] = False; rec["err"] = "out of NYC scope"
+            return "sandy", None, rec
+        pt_geom = (gpd.GeoDataFrame(geometry=[Point(lon, lat)], crs="EPSG:4326")
+                   .to_crs("EPSG:2263").iloc[0].geometry)
+        flag = sandy_inundation.inside_raster(pt_geom)
+        rec["ok"] = True; rec["result"] = {"inside": flag}
+        return "sandy", flag, rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("sandy failed")
+        return "sandy", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+def _run_dep(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("dep_stormwater")
+    try:
+        if not _in_nyc(lat, lon):
+            rec["ok"] = False; rec["err"] = "out of NYC scope"
+            return "dep", None, rec
+        pt_geom = (gpd.GeoDataFrame(geometry=[Point(lon, lat)], crs="EPSG:4326")
+                   .to_crs("EPSG:2263").iloc[0].geometry)
+        out: dict[str, Any] = {}
+        for scen in ["dep_extreme_2080", "dep_moderate_2050", "dep_moderate_current"]:
+            cls = dep_stormwater.join_raster(pt_geom, scen)
+            out[scen] = {
+                "depth_class": cls,
+                "depth_label": dep_stormwater.DEPTH_CLASS.get(cls, "outside"),
+                "citation": f"NYC DEP Stormwater Flood Map — {dep_stormwater.label(scen)}",
+            }
+        rec["ok"] = True; rec["result"] = {k: v["depth_label"] for k, v in out.items()}
+        return "dep", out, rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("dep failed")
+        return "dep", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+def _run_floodnet(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("floodnet")
+    try:
+        if not _in_nyc(lat, lon):
+            rec["ok"] = False; rec["err"] = "out of NYC scope"
+            return "floodnet", None, rec
+        s = floodnet.summary_for_point(lat, lon, radius_m=600)
+        s["radius_m"] = 600
+        rec["ok"] = True
+        rec["result"] = {"n_sensors": s["n_sensors"], "n_events_3y": s["n_flood_events_3y"]}
+        return "floodnet", s, rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("floodnet failed")
+        return "floodnet", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+def _run_311(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("nyc311")
+    try:
+        if not _in_nyc(lat, lon):
+            rec["ok"] = False; rec["err"] = "out of NYC scope"
+            return "nyc311", None, rec
+        s = nyc311.summary_for_point(lat, lon, radius_m=200, years=5)
+        rec["ok"] = True; rec["result"] = {"n": s["n"]}
+        return "nyc311", s, rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("311 failed")
+        return "nyc311", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+def _run_ida_hwm(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("ida_hwm_2021")
+    try:
+        s = ida_hwm.summary_for_point(lat, lon, radius_m=800)
+        if s is None:
+            rec["ok"] = False; rec["err"] = "HWM data missing"
+            return "ida_hwm", None, rec
+        rec["ok"] = True
+        rec["result"] = {
+            "n_within_800m": s.n_within_radius,
+            "max_height_above_gnd_ft": s.max_height_above_gnd_ft,
+            "nearest_m": s.nearest_dist_m,
+        }
+        return "ida_hwm", vars(s), rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("ida_hwm failed")
+        return "ida_hwm", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+def _run_prithvi(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("prithvi_eo_v2")
+    try:
+        if not _in_nyc(lat, lon):
+            rec["ok"] = False; rec["err"] = "out of NYC scope"
+            return "prithvi_water", None, rec
+        s = prithvi_water.summary_for_point(lat, lon)
+        if s is None:
+            rec["ok"] = False; rec["err"] = "Prithvi mask missing"
+            return "prithvi_water", None, rec
+        rec["ok"] = True
+        rec["result"] = {
+            "inside_water_polygon": s.inside_water_polygon,
+            "nearest_distance_m": s.nearest_distance_m,
+            "n_polygons_within_500m": s.n_polygons_within_500m,
+        }
+        return "prithvi_water", vars(s), rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("prithvi failed")
+        return "prithvi_water", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+def _run_microtopo(lat, lon) -> tuple[str, Any, dict]:
+    rec = _make_rec("microtopo_lidar")
+    try:
+        if not _in_nyc(lat, lon):
+            rec["ok"] = False; rec["err"] = "out of NYC scope"
+            return "microtopo", None, rec
+        m = microtopo.microtopo_at(lat, lon)
+        if m is None:
+            rec["ok"] = False; rec["err"] = "DEM fetch failed"
+            return "microtopo", None, rec
+        rec["ok"] = True
+        rec["result"] = {
+            "elev_m": m.point_elev_m,
+            "pct_200m": m.rel_elev_pct_200m,
+            "relief_m": m.basin_relief_m,
+        }
+        return "microtopo", vars(m), rec
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("microtopo failed")
+        return "microtopo", None, rec
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
+_CORNERSTONE_WORKERS = [
+    _run_sandy, _run_dep, _run_floodnet, _run_311,
+    _run_ida_hwm, _run_prithvi, _run_microtopo,
+]
+
+
+@action(reads=["lat", "lon"],
+        writes=["sandy", "dep", "floodnet", "nyc311",
+                "ida_hwm", "prithvi_water", "microtopo", "trace"])
+def step_cornerstone(state: State) -> State:
+    """Run all 7 geospatial Cornerstone specialists in parallel.
+
+    Uses ThreadPoolExecutor internally (not Burr's parallel executor) to
+    avoid the post-action cleanup hang that occurred with the previous
+    fan-out approach. Workers are pure functions — no shared Burr state."""
+    trace = list(state.get("trace", []))
+    lat, lon = state.get("lat"), state.get("lon")
+
+    defaults = {
+        "sandy": None, "dep": None, "floodnet": None,
+        "nyc311": None, "ida_hwm": None, "prithvi_water": None, "microtopo": None,
+    }
+
+    if lat is None:
+        for fn in _CORNERSTONE_WORKERS:
+            rec = _make_rec(fn.__name__.removeprefix("_run_"))
+            rec["ok"] = False; rec["err"] = "no coords"
+            rec["elapsed_s"] = 0.0
+            trace.append(rec)
+        return state.update(**defaults, trace=trace)
+
+    results: dict[str, Any] = {}
+    for fn in _CORNERSTONE_WORKERS:
+        try:
+            key, val, rec = fn(lat, lon)
+        except Exception as e:
+            rec = {"step": fn.__name__, "ok": False,
+                   "err": str(e), "elapsed_s": 0.0, "started_at": time.time()}
+            key = fn.__name__.removeprefix("_run_")
+            val = None
+            log.exception("cornerstone worker %s raised", fn.__name__)
+        results[key] = val
+        trace.append(rec)
+
+    return state.update(
+        sandy=results.get("sandy"),
+        dep=results.get("dep"),
+        floodnet=results.get("floodnet"),
+        nyc311=results.get("nyc311"),
+        ida_hwm=results.get("ida_hwm"),
+        prithvi_water=results.get("prithvi_water"),
+        microtopo=results.get("microtopo"),
+        trace=trace,
+    )
+
+
 @action(reads=["query"], writes=["geocode", "lat", "lon", "trace"])
 def step_geocode(state: State) -> State:
     rec, trace = _step(state, "geocode")
@@ -601,6 +836,28 @@ def step_floodnet_forecast(state: State) -> State:
         rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
 
 
+@action(reads=["lat", "lon"], writes=["npcc4_slr", "trace"])
+def step_npcc4_projection(state: State) -> State:
+    """NPCC4 (2024) sea-level rise table — static lookup, always available."""
+    rec, trace = _step(state, "npcc4_projection")
+    try:
+        s = npcc4_slr.get_projections()
+        rec["ok"] = True
+        rec["result"] = {
+            "2050_10th_in": s["2050"]["10"]["in"],
+            "2050_50th_in": s["2050"]["50"]["in"],
+            "2050_90th_in": s["2050"]["90"]["in"],
+            "2100_90th_in": s["2100"]["90"]["in"],
+        }
+        return state.update(npcc4_slr=s, trace=trace)
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("npcc4_projection failed")
+        return state.update(npcc4_slr=None, trace=trace)
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
 @action(reads=["lat", "lon"], writes=["mta_entrances", "trace"])
 def step_mta_entrances(state: State) -> State:
     rec, trace = _step(state, "mta_entrance_exposure")
@@ -961,11 +1218,12 @@ def _label_counts(gliner_out: dict[str, dict]) -> dict[str, int]:
                "ida_hwm", "prithvi_water", "prithvi_live", "terramind",
                "terramind_lulc", "terramind_buildings",
                "noaa_tides", "nws_alerts", "nws_obs", "ttm_forecast",
-               "ttm_311_forecast", "floodnet_forecast", "ttm_battery_surge",
+               "ttm_311_forecast", "floodnet_forecast", "npcc4_slr",
+               "ttm_battery_surge",
                "mta_entrances",
                "nycha_developments", "doe_schools", "doh_hospitals",
                "rag", "gliner"],
-        writes=["paragraph", "audit", "mellea", "trace"])
+        writes=["paragraph", "audit", "mellea", "citations", "trace"])
 def step_reconcile(state: State) -> State:
     is_strict = _current_strict_mode()
     rec, trace = _step(state, "mellea_reconcile_address" if is_strict else "reconcile_granite41")
@@ -986,6 +1244,7 @@ def step_reconcile(state: State) -> State:
             "ttm_forecast": state.get("ttm_forecast"),
             "ttm_311_forecast": state.get("ttm_311_forecast"),
             "floodnet_forecast": state.get("floodnet_forecast"),
+            "npcc4_slr": state.get("npcc4_slr"),
             "ttm_battery_surge": state.get("ttm_battery_surge"),
             "rag": state.get("rag"),
             "gliner": state.get("gliner"),
@@ -1010,8 +1269,19 @@ def step_reconcile(state: State) -> State:
             else:
                 token_cb = _current_token_callback()
                 attempt_cb = _current_mellea_attempt_callback()
+                # Enumerate the exact doc_ids the model may cite so it
+                # doesn't invent plausible-sounding ones (e.g. rag_npcc4).
+                _avail_ids = sorted(
+                    m["role"].split(" ", 1)[1]
+                    for m in doc_msgs
+                    if m.get("role", "").startswith("document ")
+                )
+                _id_note = (
+                    f"\nValid document IDs for citation (use these exactly): "
+                    f"{', '.join(_avail_ids)}."
+                )
                 framed_prompt = augment_system_prompt(
-                    EXTRA_SYSTEM_PROMPT,
+                    EXTRA_SYSTEM_PROMPT + _id_note,
                     query=_current_user_query() or state.get("query") or "",
                     intent=_current_planner_intent() or "single_address",
                 )
@@ -1046,6 +1316,13 @@ def step_reconcile(state: State) -> State:
                     "model": mres["model"],
                     "loop_budget": mres["loop_budget"],
                 }
+                # If Mellea returned empty (streaming stall / LLM failure),
+                # do NOT call run_reconcile as a fallback: Mellea's daemon
+                # thread is likely still running a streaming vLLM request,
+                # and a second concurrent request overloads RunPod, causing
+                # both to hang for the full 240 s LiteLLM timeout.
+                if not para or len(para.strip()) < 50:
+                    log.warning("mellea returned empty — skipping fallback to avoid concurrent vLLM")
             rec["result"] = {
                 "rerolls": (mellea_meta or {}).get("rerolls"),
                 "passed": (f"{len((mellea_meta or {}).get('requirements_passed') or [])}/"
@@ -1059,14 +1336,19 @@ def step_reconcile(state: State) -> State:
                 "paragraph_chars": len(para),
                 "dropped_sentences": len(audit["dropped"]),
             }
+        # Build citation metadata list from whichever doc_msgs were used.
+        from app.reconcile import build_documents, trim_docs_to_plan
+        _cite_msgs = build_documents(snap)
+        _cite_msgs = trim_docs_to_plan(_cite_msgs, _current_planned_specialists())
+        cite_list = citations_from_docs(_cite_msgs)
         rec["ok"] = True
         return state.update(paragraph=para, audit=audit,
-                            mellea=mellea_meta, trace=trace)
+                            mellea=mellea_meta, citations=cite_list, trace=trace)
     except Exception as e:
         rec["ok"] = False; rec["err"] = str(e)
         log.exception("reconcile failed")
         return state.update(paragraph="", audit={"raw": "", "dropped": []},
-                            mellea=None, trace=trace)
+                            mellea=None, citations=[], trace=trace)
     finally:
         rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
 
@@ -1117,37 +1399,45 @@ _NYCHA_REGISTERS_ENABLED = _os.environ.get(
 ).lower() in ("1", "true", "yes")
 
 
-def build_app(query: str):
-    """Linear, single-action-per-step Burr application.
+_BURR_TRACKING_DIR = _os.environ.get("RIPRAP_BURR_TRACKING_DIR", "/tmp/riprap-burr")
 
-    Order: cheap-first geo + flood layers, then live live network signals,
-    then RAG → reconcile. Heavy specialists (NYCHA / DOE / DOH register
-    joins, Prithvi-EO live STAC, TerraMind diffusion) are gated behind
-    RIPRAP_HEAVY_SPECIALISTS — see the module-level note above.
+
+def build_app(query: str, step_queue=None):
+    """Burr application — Cornerstone specialists run in parallel.
+
+    Order: geocode → cornerstone (7 geospatial specialists, parallel) →
+    live network signals → RAG → reconcile. Heavy specialists (NYCHA /
+    DOE / DOH register joins, Prithvi-EO live STAC, TerraMind diffusion)
+    are gated behind RIPRAP_HEAVY_SPECIALISTS — see module-level note.
+
+    step_queue: optional queue.Queue — if provided, StepEventHook pushes
+    each completed action's trace record to it (replaces iter_steps
+    manual deduplication). LocalTrackingClient writes to RIPRAP_BURR_TRACKING_DIR.
+    SQLitePersister caches completed runs keyed by (address, date) so repeat
+    queries skip the specialist pipeline and go straight to reconcile.
     """
+    tracker = LocalTrackingClient(project="riprap", storage_dir=_BURR_TRACKING_DIR)
+
     builder = (
         ApplicationBuilder()
         .with_state(query=query, trace=[])
         .with_entrypoint("geocode")
+        .with_tracker(tracker)
+        .with_hooks(StepEventHook(step_queue))
     )
 
     actions: dict[str, Any] = {
         "geocode": step_geocode,
-        "sandy": step_sandy,
-        "dep": step_dep,
-        "floodnet": step_floodnet,
-        "nyc311": step_311,
+        "cornerstone": step_cornerstone,  # sandy+dep+floodnet+311+ida+prithvi+microtopo
         "noaa_tides": step_noaa_tides,
         "nws_alerts": step_nws_alerts,
         "nws_obs": step_nws_obs,
         "ttm_forecast": step_ttm_forecast,
         "ttm_311_forecast": step_ttm_311_forecast,
         "floodnet_forecast": step_floodnet_forecast,
+        "npcc4_projection": step_npcc4_projection,
         "ttm_battery_surge": step_ttm_battery_surge,
-        "microtopo": step_microtopo,
-        "ida_hwm": step_ida_hwm,
         "mta_entrances": step_mta_entrances,
-        "prithvi": step_prithvi,  # baked GeoJSON polygons for Ida; cheap
     }
     if _HEAVY_SPECIALISTS_ENABLED and _NYCHA_REGISTERS_ENABLED:
         actions["nycha"] = step_nycha
@@ -1156,10 +1446,6 @@ def build_app(query: str):
     if _HEAVY_SPECIALISTS_ENABLED:
         actions["prithvi_live"] = step_prithvi_live
         actions["terramind"] = step_terramind
-        # New TerraMind-NYC LoRA family — one chip fetch feeds two
-        # specialists. Keep eo_chip directly before the two consumers
-        # so the chip stays warm in memory and isn't garbage-collected
-        # by anything in between.
         actions["eo_chip"] = step_eo_chip
         actions["terramind_lulc"] = step_terramind_lulc
         actions["terramind_buildings"] = step_terramind_buildings
@@ -1167,9 +1453,18 @@ def build_app(query: str):
     actions["gliner"] = step_gliner
     actions["reconcile"] = step_reconcile
 
-    # Sequential transitions — pair every adjacent action in the dict order.
+    # Conditional transitions:
+    #   geocode → cornerstone if coords resolved; else skip straight to reconcile
+    #   All other transitions remain sequential.
     keys = list(actions.keys())
-    transitions = list(zip(keys, keys[1:]))
+    # Build sequential pairs, but replace geocode→cornerstone with a conditional.
+    transitions = []
+    for src, dst in zip(keys, keys[1:]):
+        if src == "geocode" and dst == "cornerstone":
+            transitions.append(("geocode", "cornerstone", expr("lat is not None")))
+            transitions.append(("geocode", "reconcile"))  # geocode failed → skip all
+        else:
+            transitions.append((src, dst))
 
     return (
         builder.with_actions(**actions).with_transitions(*transitions).build()
@@ -1252,16 +1547,7 @@ def iter_steps(query: str):
     import queue
 
     q: queue.Queue[tuple[str, Any] | None] = queue.Queue()
-    seen_keys: set[tuple[str, float]] = set()
-
-    def _push_step(rec: dict) -> None:
-        key = (rec.get("step", ""), rec.get("started_at", 0.0))
-        if key in seen_keys:
-            return
-        seen_keys.add(key)
-        q.put(("step", rec))
-
-    app = build_app(query)
+    app = build_app(query, step_queue=q)
     final_state_holder: dict[str, Any] = {}
 
     # Threadlocals are per-thread; the request thread (single_address.run
@@ -1284,12 +1570,8 @@ def iter_steps(query: str):
         try:
             for _action_obj, _result, state in app.iterate(halt_after=["reconcile"]):
                 final_state_holder["state"] = state
-                # Each action appends one record to state.trace; emit the
-                # most recent so the SSE client gets the step event the
-                # moment Burr returns from that action.
-                trace = state.get("trace") or []
-                if trace:
-                    _push_step(trace[-1])
+                # StepEventHook fires after each action and pushes to q;
+                # nothing else needed here.
         except Exception as e:
             log.exception("iterate raised")
             q.put(("error", {"err": f"{type(e).__name__}: {e}"}))
@@ -1358,6 +1640,7 @@ def iter_steps(query: str):
         "paragraph": state.get("paragraph"),
         "audit": state.get("audit"),
         "mellea": state.get("mellea"),
+        "citations": state.get("citations"),
         "energy": _summarize_energy(trace),
         "emissions": _summarize_emissions(),
     }

app/llm.py

@@ -92,6 +92,15 @@ def _build_router() -> Router:
     fallbacks: list[dict[str, list[str]]] = []
     use_vllm = _PRIMARY == "vllm" and bool(_VLLM_BASE)
 
+    # vLLM on RunPod can take 250+ seconds to cold-start (container boot +
+    # model load into GPU VRAM). The first-token timeout must exceed that.
+    # stream_timeout (per-chunk) stays tight since subsequent tokens are fast.
+    _vllm_first_token_timeout = int(
+        os.environ.get("RIPRAP_LITELLM_TIMEOUT_S", "360"))
+    # 5s: fail fast so callers (mellea probe loop) aren't blocked waiting
+    # for an Ollama that doesn't exist in the vLLM-primary HF Space.
+    _ollama_timeout = 5
+
     for alias, (vllm_name, ollama_tag) in _LOGICAL.items():
         if use_vllm:
             model_list.append({
@@ -100,8 +109,8 @@ def _build_router() -> Router:
                     "model": f"openai/{vllm_name}",
                     "api_base": _VLLM_BASE,
                     "api_key": _VLLM_KEY,
-                    "timeout": 240,
-                    "stream_timeout": 240,
+                    "timeout": _vllm_first_token_timeout,
+                    "stream_timeout": 60,
                 },
             })
             if _FALLBACK == "ollama":
@@ -111,8 +120,8 @@ def _build_router() -> Router:
                     "litellm_params": {
                         "model": f"ollama_chat/{ollama_tag}",
                         "api_base": _OLLAMA_BASE,
-                        "timeout": 240,
-                        "stream_timeout": 240,
+                        "timeout": _ollama_timeout,
+                        "stream_timeout": _ollama_timeout,
                     },
                 })
                 fallbacks.append({alias: [fb_alias]})
@@ -122,8 +131,8 @@ def _build_router() -> Router:
                 "litellm_params": {
                     "model": f"ollama_chat/{ollama_tag}",
                     "api_base": _OLLAMA_BASE,
-                    "timeout": 240,
-                    "stream_timeout": 240,
+                    "timeout": _ollama_timeout,
+                    "stream_timeout": _ollama_timeout,
                 },
             })
 
@@ -135,7 +144,7 @@ def _build_router() -> Router:
         fallbacks=fallbacks,
         num_retries=0,  # Router fallback handles the failover; no point
                         # burning seconds re-hitting a dead endpoint.
-        timeout=240,
+        timeout=_vllm_first_token_timeout if use_vllm else _ollama_timeout,
     )
 
 

app/mellea_validator.py

@@ -23,7 +23,9 @@ from __future__ import annotations
 
 import logging
 import os
+import queue
 import re
+import threading
 import time
 from typing import Any
 
@@ -271,7 +273,7 @@ def reconcile_strict(doc_msgs: list[dict],
             return_sampling_results=True,
             model_options={"temperature": 0,
                            "num_ctx": int(os.environ.get("RIPRAP_MELLEA_NUM_CTX", "4096")),
-                           "num_predict": int(os.environ.get("RIPRAP_MELLEA_NUM_PREDICT", "400")),
+                           "num_predict": int(os.environ.get("RIPRAP_MELLEA_NUM_PREDICT", "600")),
                            **(ollama_options or {})},
         )
 
@@ -339,6 +341,8 @@ def reconcile_strict_streaming(
     t0 = time.time()
 
     checks = [
+        ("non_empty",
+         lambda p: bool(p and len(p.strip()) > 50)),
         ("numerics_grounded",
          _check_no_invented_numbers(doc_msgs)),
         ("no_placeholder_tokens",
@@ -353,23 +357,61 @@ def reconcile_strict_streaming(
         {"role": "system", "content": system_prompt},
         {"role": "user", "content": user_prompt},
     ]
-    # num_ctx 4096 fits a typical trimmed prompt (≈700 system + ≈2500 docs);
-    # num_predict 400 caps the 4-section briefing at ≈300-350 tokens. With
-    # RIPRAP_TRIM_DOCS=1 and the planner picking 6-9 specialists, the 4096
-    # window has been sufficient on every probe; the previous 6144/600 was
-    # sized for the *untrimmed* fan-out and was forcing Ollama to grow the
-    # KV cache (33% more memory + a full re-init) every Mellea attempt.
-    # Override with RIPRAP_MELLEA_NUM_CTX / RIPRAP_MELLEA_NUM_PREDICT.
+    # num_predict 512 lets the 4-section briefing complete in one pass.
+    # Reconciler prompts run ~1200 tokens (after trim_docs_to_plan),
+    # so 1200+512=1712 comfortably under the vLLM max_model_len=2352.
+    # Override with RIPRAP_MELLEA_NUM_PREDICT if needed.
+    # num_ctx (Ollama only) is forwarded via extra_body; vLLM ignores it.
     base_opts = {"temperature": 0,
                  "num_ctx": int(os.environ.get("RIPRAP_MELLEA_NUM_CTX", "4096")),
-                 "num_predict": int(os.environ.get("RIPRAP_MELLEA_NUM_PREDICT", "400")),
+                 "num_predict": int(os.environ.get("RIPRAP_MELLEA_NUM_PREDICT", "512")),
                  **(ollama_options or {})}
 
     paragraph = ""
     last_passed: list[str] = []
     last_failed: list[str] = [name for name, _ in checks]
     last_paragraph = ""
+    best_paragraph = ""  # best non-empty paragraph seen across all attempts
     attempts = 0
+    _streaming_hung = False  # set on first per-token timeout; skip retries
+
+    # Two-phase timeout: the FIRST token from a cold RunPod pod can take
+    # 3-4 min (container boot + model load into GPU VRAM). Once streaming
+    # has started, each subsequent token should arrive in < 5 s; we use a
+    # tight 45 s inter-token timeout to catch mid-stream stalls quickly.
+    _first_token_timeout = int(os.environ.get("RIPRAP_FIRST_TOKEN_TIMEOUT_S", "400"))
+    _inter_token_timeout = int(os.environ.get("RIPRAP_TOKEN_TIMEOUT_S", "45"))
+
+    # When PRIMARY=vllm, RunPod cold-starts take ~250s (container boot +
+    # model load). LiteLLM gets a 503 and falls back to Ollama, which blocks
+    # for the full Ollama timeout before failing. Poll /v1/models instead and
+    # wait here — keepalives keep the SSE connection alive during the wait.
+    _vllm_base = os.environ.get("RIPRAP_LLM_BASE_URL", "").rstrip("/")
+    if os.environ.get("RIPRAP_LLM_PRIMARY", "ollama") == "vllm" and _vllm_base:
+        try:
+            import httpx as _httpx
+            _probe_url = f"{_vllm_base}/models"
+            _probe_key = os.environ.get("RIPRAP_LLM_API_KEY", "") or "EMPTY"
+            _probe_headers = {"Authorization": f"Bearer {_probe_key}"}
+            _probe_deadline = t0 + _first_token_timeout
+            log.info("mellea: polling vLLM readiness at %s", _probe_url)
+            while time.time() < _probe_deadline:
+                try:
+                    _r = _httpx.get(_probe_url, headers=_probe_headers, timeout=5.0)
+                    # Any non-503/502/504 means the service is UP (200 = ready,
+                    # 401 = auth-gated but alive, 404 = wrong path but alive).
+                    if _r.status_code not in (502, 503, 504):
+                        log.info("mellea: vLLM ready (status=%d, %.1fs elapsed)",
+                                 _r.status_code, time.time() - t0)
+                        break
+                except Exception as _pe:
+                    log.debug("mellea: vLLM probe: %r", _pe)
+                time.sleep(10)
+            else:
+                log.warning("mellea: vLLM not ready after %.1fs, proceeding anyway",
+                            time.time() - t0)
+        except ImportError:
+            log.warning("mellea: httpx not available, skipping vLLM probe")
 
     for attempt_idx in range(loop_budget):
         attempts = attempt_idx + 1
@@ -401,10 +443,52 @@ def reconcile_strict_streaming(
             messages.append({"role": "user", "content": "\n".join(feedback)})
 
         chunks: list[str] = []
-        for chunk in llm.chat(model=model, messages=messages,
-                                 stream=True, options=base_opts):
+
+        # Each attempt gets its own sentinel so that a stale daemon thread
+        # from a previous timed-out attempt cannot corrupt this attempt's
+        # queue (the closure captures variables by reference; re-binding
+        # them per-attempt keeps each daemon's sentinel unique).
+        _stream_q: queue.Queue = queue.Queue()
+        _done_sentinel = object()
+
+        def _stream_worker(q=_stream_q, done=_done_sentinel,
+                           msgs=messages, opts=base_opts):
+            try:
+                for _chunk in llm.chat(model=model, messages=msgs,
+                                       stream=True, options=opts):
+                    q.put(_chunk)
+            except Exception as _e:
+                q.put(_e)
+            finally:
+                q.put(done)
+
+        _st = threading.Thread(target=_stream_worker, daemon=True)
+        _st.start()
+        _timed_out = False
+        _got_first_token = False
+        while True:
+            _timeout = _inter_token_timeout if _got_first_token else _first_token_timeout
+            try:
+                chunk = _stream_q.get(timeout=_timeout)
+            except queue.Empty:
+                log.warning("mellea: timeout (%ds, first=%s) — breaking stream",
+                            _timeout, not _got_first_token)
+                _timed_out = True
+                _streaming_hung = True
+                break
+            if chunk is _done_sentinel:
+                break
+            if isinstance(chunk, Exception):
+                log.warning("mellea: stream error: %r", chunk)
+                if not _got_first_token:
+                    # LiteLLM/httpx timeout before first token — treat as
+                    # streaming hung so we don't start a concurrent retry.
+                    _timed_out = True
+                    _streaming_hung = True
+                break
             delta = (chunk.get("message") or {}).get("content") or ""
             if delta:
+                _got_first_token = True
                 chunks.append(delta)
                 if on_token is not None:
                     try:
@@ -412,6 +496,8 @@ def reconcile_strict_streaming(
                     except Exception:
                         log.exception("on_token callback raised")
         paragraph = "".join(chunks).strip()
+        if paragraph:
+            best_paragraph = paragraph
 
         passed: list[str] = []
         failed: list[str] = []
@@ -434,8 +520,18 @@ def reconcile_strict_streaming(
         if not failed:
             break
 
+        # If this attempt's stream hung, stop retrying with streaming.
+        # A stale daemon thread is still consuming vLLM resources; starting
+        # another streaming request would create a second concurrent request
+        # and can crash vLLM (observed as HTTP/2 stream error on the SSE
+        # connection). Signal the caller to use a non-streaming fallback.
+        if _timed_out:
+            log.warning("mellea: streaming hung — aborting retry loop "
+                        "to avoid concurrent vLLM requests")
+            break
+
     return {
-        "paragraph": paragraph,
+        "paragraph": paragraph or best_paragraph,
         "rerolls": max(0, attempts - 1),
         "n_attempts": attempts,
         "requirements_total": len(checks),

web/main.py

@@ -613,13 +613,26 @@ def _run_compare(p, raw_query: str, out_q, i_addr) -> dict:
 
     mellea_a = results[0][2].get("mellea") or {}
     mellea_b = results[1][2].get("mellea") or {}
-    return {
+
+    # Spread Place A's full specialist state into the return dict so
+    # adaptFinalToFindings can build evidence cards (TTM, TerraMind, Prithvi,
+    # Sandy, etc.) from the higher-risk location.  Place B's live-state data
+    # is available via targets[].state for future per-location card rendering.
+    # Without this, _run_compare returned only paragraph/mellea/intent/targets
+    # and all fine-tuned model cards were silently suppressed (state keys
+    # missing → card builders returned null).
+    out = {**results[0][2]}
+    out.update({
         "paragraph": merged_paragraph,
         "mellea": _merge_mellea(mellea_a, mellea_b),
         "intent": "compare",
-        "targets": [{"label": lbl, "address": addr} for lbl, addr, _ in results],
+        "targets": [
+            {"label": lbl, "address": addr, "state": res}
+            for lbl, addr, res in results
+        ],
         "tier": results[0][2].get("tier"),
-    }
+    })
+    return out
 
 
 @app.get("/api/agent")
@@ -682,6 +695,21 @@ async def api_agent_stream(q: str):
     def runner():
         emissions.install(tracker)
         try:
+            import threading as _th
+            from app import llm as _llm
+
+            def _warmup_llm():
+                try:
+                    _llm.chat(
+                        model="granite-8b",
+                        messages=[{"role": "user", "content": "hi"}],
+                        options={"num_predict": 1, "temperature": 0},
+                        stream=False,
+                    )
+                except Exception:
+                    pass
+            _th.Thread(target=_warmup_llm, daemon=True, name="riprap-warmup").start()
+
             from app.intents import development_check as i_dev
             from app.intents import live_now as i_live
             from app.intents import neighborhood as i_nbhd
@@ -752,6 +780,10 @@ async def api_agent_stream(q: str):
             try:
                 ev = await asyncio.to_thread(out_q.get, True, 1.0)
             except Exception:
+                # No event for 1 s — send an SSE comment so the HF Space
+                # proxy doesn't close the idle connection (proxy idle timeout
+                # is ~15-20 s; the reconciler's vLLM call can take longer).
+                yield ": keepalive\n\n"
                 continue
             kind = ev.get("kind")
             if kind == "_done":