Stones C7: emit stone_start / stone_done SSE envelope

Wraps the FSM step stream in a Stone-boundary envelope so the trace
UI can render the four data-Stones + Capstone as collapsible parent
nodes around their atomic specialist calls.

web/main.py
Adds _STEP_TO_STONE mapping FSM step names to Stone groups, plus
_STONE_META (NAME / TAGLINE / DESCRIPTION) drawn straight from the
app/stones modules so taglines stay single-source.
The /api/agent/stream event_stream() now tracks the current Stone
and emits:
- stone_start when the first step of a new group fires (or when
the first reconcile token arrives, opening Capstone before the
FSM's reconcile step event lands at the end);
- stone_done when the next group opens, the pipeline completes
(final), or it terminates without a final.
Ancillary steps (geocode, rag_granite_embedding, gliner_extract,
nta_resolve) flow through without opening a Stone — they're
orientation / policy infrastructure shared across Stones.

web/static/agent.js
Registers stone_start / stone_done listeners that buffer the events
in TRACE_BUF for the auditable report and call optional
markStoneStart / markStoneDone on the <r-trace> element if the
Svelte component implements them. The collapsible parent-row UI
is a follow-up Svelte rebuild; the SSE backend lands today so the
component can be regenerated against the new event vocabulary.

tests/test_stone_envelope.py
Pure-Python replay of the envelope state machine. Covers:
- Stone-mapping coverage (every expected step name is in main.py)
- Full pipeline open/close ordering (Cornerstone -> Keystone ->
Touchstone -> Lodestone -> Capstone)
- Ancillary steps don't open a Stone
- tokens-before-step opens Capstone exactly once
- Premature termination closes any open Stone

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

tests/test_stone_envelope.py

@@ -0,0 +1,165 @@
+"""Verify the SSE stone_start / stone_done envelope around step events.
+
+The web layer wraps the FSM step stream and emits stone_start / stone_done
+events at Stone-group boundaries. We don't need a running server to test
+the boundary logic — we extract the iterating segment of api_agent_stream
+into a pure generator and feed it a synthetic event sequence.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+# main.py does heavy work at import (warm caches, mount FastAPI routes);
+# we only need its module-level constants. The cheapest path is to
+# re-derive _STEP_TO_STONE the same way main.py does — the dict is
+# tiny — and assert structurally that main.py still references each
+# expected step name, so a future drift fails this test.
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_step_to_stone_mapping_covers_known_steps():
+    src = (ROOT / "web" / "main.py").read_text()
+    # Smoke: every step name we expect is in the mapping.
+    for step in ("sandy_inundation", "dep_stormwater", "ida_hwm_2021",
+                 "prithvi_eo_v2", "microtopo_lidar",
+                 "mta_entrance_exposure", "nycha_development_exposure",
+                 "doe_school_exposure", "doh_hospital_exposure",
+                 "terramind_synthesis", "eo_chip_fetch",
+                 "terramind_buildings",
+                 "floodnet", "nyc311", "nws_obs", "noaa_tides",
+                 "prithvi_eo_live", "terramind_lulc",
+                 "nws_alerts", "ttm_forecast", "ttm_311_forecast",
+                 "floodnet_forecast", "ttm_battery_surge",
+                 "reconcile_granite41", "mellea_reconcile_address"):
+        assert f'"{step}"' in src, f"web/main.py missing step mapping {step!r}"
+
+
+def _replay(events: list[dict]) -> list[tuple[str, dict]]:
+    """Local re-implementation of the SSE envelope state machine in
+    web/main.py:event_stream. We only emulate the parts that touch
+    stone_start / stone_done / step / token / final."""
+    # Match web/main.py exactly.
+    STEP_TO_STONE = {
+        "sandy_inundation": "Cornerstone", "dep_stormwater": "Cornerstone",
+        "ida_hwm_2021": "Cornerstone", "prithvi_eo_v2": "Cornerstone",
+        "microtopo_lidar": "Cornerstone",
+        "mta_entrance_exposure": "Keystone",
+        "nycha_development_exposure": "Keystone",
+        "doe_school_exposure": "Keystone",
+        "doh_hospital_exposure": "Keystone",
+        "terramind_synthesis": "Keystone",
+        "eo_chip_fetch": "Keystone",
+        "terramind_buildings": "Keystone",
+        "floodnet": "Touchstone", "nyc311": "Touchstone",
+        "nws_obs": "Touchstone", "noaa_tides": "Touchstone",
+        "prithvi_eo_live": "Touchstone", "terramind_lulc": "Touchstone",
+        "nws_alerts": "Lodestone", "ttm_forecast": "Lodestone",
+        "ttm_311_forecast": "Lodestone", "floodnet_forecast": "Lodestone",
+        "ttm_battery_surge": "Lodestone",
+        "reconcile_granite41": "Capstone",
+        "mellea_reconcile_address": "Capstone",
+    }
+    out: list[tuple[str, dict]] = []
+    current = None
+
+    def open_(s):
+        out.append(("stone_start", {"name": s}))
+
+    def close_(s):
+        out.append(("stone_done", {"name": s}))
+
+    for ev in events:
+        kind = ev["kind"]
+        if kind == "token" and current != "Capstone":
+            if current is not None:
+                close_(current)
+            current = "Capstone"
+            open_(current)
+        if kind == "step":
+            stone = STEP_TO_STONE.get(ev["step"])
+            if stone is not None and stone != current:
+                if current is not None:
+                    close_(current)
+                current = stone
+                open_(current)
+        if kind == "final" and current is not None:
+            close_(current)
+            current = None
+        out.append((kind, ev))
+    if current is not None:
+        close_(current)
+    return out
+
+
+def _names(seq, kind):
+    return [d["name"] for k, d in seq if k == kind]
+
+
+def test_envelope_around_full_pipeline():
+    events = [
+        {"kind": "step", "step": "geocode"},        # not in any Stone
+        {"kind": "step", "step": "sandy_inundation"},
+        {"kind": "step", "step": "dep_stormwater"},
+        {"kind": "step", "step": "mta_entrance_exposure"},
+        {"kind": "step", "step": "floodnet"},
+        {"kind": "step", "step": "nyc311"},
+        {"kind": "step", "step": "ttm_forecast"},
+        {"kind": "step", "step": "ttm_battery_surge"},
+        {"kind": "step", "step": "rag_granite_embedding"},  # ancillary
+        {"kind": "token", "delta": "**Status**"},
+        {"kind": "final", "paragraph": "..."},
+    ]
+    out = _replay(events)
+    starts = _names(out, "stone_start")
+    dones = _names(out, "stone_done")
+    assert starts == ["Cornerstone", "Keystone", "Touchstone",
+                       "Lodestone", "Capstone"]
+    assert dones == ["Cornerstone", "Keystone", "Touchstone",
+                      "Lodestone", "Capstone"]
+
+
+def test_envelope_skips_ancillary_steps_cleanly():
+    """geocode / rag / gliner aren't part of any Stone — they shouldn't
+    open or close a Stone boundary."""
+    events = [
+        {"kind": "step", "step": "geocode"},
+        {"kind": "step", "step": "rag_granite_embedding"},
+        {"kind": "step", "step": "gliner_extract"},
+    ]
+    out = _replay(events)
+    assert _names(out, "stone_start") == []
+    assert _names(out, "stone_done") == []
+
+
+def test_envelope_handles_token_before_capstone_step():
+    """Reconcile streams tokens BEFORE the FSM emits a step event for
+    reconcile (the step fires on completion). The envelope must
+    open Capstone on the first token, not wait for the step."""
+    events = [
+        {"kind": "step", "step": "sandy_inundation"},
+        {"kind": "token", "delta": "x"},
+        {"kind": "token", "delta": "y"},
+        {"kind": "step", "step": "reconcile_granite41"},
+        {"kind": "final", "paragraph": "..."},
+    ]
+    out = _replay(events)
+    starts = _names(out, "stone_start")
+    dones = _names(out, "stone_done")
+    assert starts == ["Cornerstone", "Capstone"]
+    assert dones == ["Cornerstone", "Capstone"]
+    # The reconcile_granite41 step shouldn't open a SECOND Capstone —
+    # it's already current.
+    assert starts.count("Capstone") == 1
+
+
+def test_envelope_closes_on_premature_end():
+    """Pipeline terminates without final (e.g. error) — any open Stone
+    must be closed so the client doesn't render an unbounded row."""
+    events = [
+        {"kind": "step", "step": "sandy_inundation"},
+        # No further events; replay() closes the open Stone.
+    ]
+    out = _replay(events)
+    assert _names(out, "stone_start") == ["Cornerstone"]
+    assert _names(out, "stone_done") == ["Cornerstone"]

web/main.py

@@ -18,6 +18,63 @@ from fastapi.staticfiles import StaticFiles  # noqa: E402
 from app.context import floodnet  # noqa: E402
 from app.flood_layers import dep_stormwater, sandy_inundation  # noqa: E402
 from app.fsm import iter_steps  # noqa: E402
+from app.stones import DATA_STONES  # noqa: E402
+from app.stones import capstone as _capstone_stone  # noqa: E402
+
+# Map FSM step name -> Stone for the SSE stone_start / stone_done envelope.
+# Steps not in this map (geocode, rag_granite_embedding, gliner_extract,
+# nta_resolve and friends) don't open a Stone boundary — they're
+# orientation / policy infrastructure shared across Stones.
+_STEP_TO_STONE: dict[str, str] = {
+    # Cornerstone
+    "sandy_inundation":           "Cornerstone",
+    "dep_stormwater":             "Cornerstone",
+    "ida_hwm_2021":               "Cornerstone",
+    "prithvi_eo_v2":              "Cornerstone",
+    "microtopo_lidar":            "Cornerstone",
+    # Keystone (the chip fetch is infrastructure for the LoRA pair, but
+    # it's logically Keystone-adjacent and we surface it under that
+    # banner so the trace doesn't show a phantom orphan step).
+    "mta_entrance_exposure":      "Keystone",
+    "nycha_development_exposure": "Keystone",
+    "doe_school_exposure":        "Keystone",
+    "doh_hospital_exposure":      "Keystone",
+    "terramind_synthesis":        "Keystone",
+    "eo_chip_fetch":              "Keystone",
+    "terramind_buildings":        "Keystone",
+    # Touchstone
+    "floodnet":                   "Touchstone",
+    "nyc311":                     "Touchstone",
+    "nws_obs":                    "Touchstone",
+    "noaa_tides":                 "Touchstone",
+    "prithvi_eo_live":            "Touchstone",
+    "terramind_lulc":             "Touchstone",
+    # Lodestone
+    "nws_alerts":                 "Lodestone",
+    "ttm_forecast":               "Lodestone",
+    "ttm_311_forecast":           "Lodestone",
+    "floodnet_forecast":          "Lodestone",
+    "ttm_battery_surge":          "Lodestone",
+    # Capstone — the reconciler step's name varies between strict and
+    # legacy paths; both map to Capstone.
+    "reconcile_granite41":        "Capstone",
+    "mellea_reconcile_address":   "Capstone",
+    "reconcile_neighborhood":     "Capstone",
+    "reconcile_development":      "Capstone",
+    "reconcile_live_now":         "Capstone",
+}
+
+# Pretty-printed Stone metadata the frontend renders as parent-row labels.
+_STONE_META: dict[str, dict] = {
+    s.NAME: {"name": s.NAME, "tagline": s.TAGLINE,
+             "description": s.DESCRIPTION}
+    for s in DATA_STONES
+}
+_STONE_META[_capstone_stone.NAME] = {
+    "name": _capstone_stone.NAME,
+    "tagline": _capstone_stone.TAGLINE,
+    "description": _capstone_stone.DESCRIPTION,
+}
 
 ROOT = Path(__file__).resolve().parent
 STATIC = ROOT / "static"
@@ -488,6 +545,27 @@ async def api_agent_stream(q: str):
         loop = asyncio.get_event_loop()
         loop.run_in_executor(None, runner)
         yield f"event: hello\ndata: {json.dumps({'query': q})}\n\n"
+
+        # Stone-boundary envelope: track current Stone so we can wrap
+        # contiguous step events in stone_start / stone_done. step
+        # events whose name maps to None (geocode, rag, gliner) flow
+        # through without opening a Stone — those are orientation /
+        # ancillary, not part of any data-Stone group.
+        current_stone: str | None = None
+        stone_step_count: dict[str, int] = {}
+
+        def _open(stone: str) -> str:
+            stone_step_count[stone] = 0
+            payload = {**_STONE_META.get(stone, {"name": stone})}
+            return f"event: stone_start\ndata: {json.dumps(payload)}\n\n"
+
+        def _close(stone: str) -> str:
+            payload = {
+                **_STONE_META.get(stone, {"name": stone}),
+                "n_steps": stone_step_count.get(stone, 0),
+            }
+            return f"event: stone_done\ndata: {json.dumps(payload)}\n\n"
+
         while True:
             try:
                 ev = await asyncio.to_thread(out_q.get, True, 1.0)
@@ -496,7 +574,45 @@ async def api_agent_stream(q: str):
             kind = ev.get("kind")
             if kind == "_done":
                 break
+
+            # First reconcile token implies the data-Stones are done
+            # and the Capstone has begun, even if the FSM step event
+            # for reconcile hasn't fired yet (it fires AFTER the
+            # generation finishes). Open Capstone here so the UI
+            # shows it lighting up while tokens stream.
+            if kind == "token" and current_stone != "Capstone":
+                if current_stone is not None:
+                    yield _close(current_stone)
+                current_stone = "Capstone"
+                yield _open(current_stone)
+
+            if kind == "step":
+                step_name = ev.get("step") or ""
+                stone = _STEP_TO_STONE.get(step_name)
+                if stone is not None:
+                    if stone != current_stone:
+                        if current_stone is not None:
+                            yield _close(current_stone)
+                        current_stone = stone
+                        yield _open(current_stone)
+                    stone_step_count[stone] = (
+                        stone_step_count.get(stone, 0) + 1)
+
+            # `final` arrives after the Capstone has produced its
+            # paragraph. Close the Capstone before forwarding final
+            # so the trace cleanly reads: ... stone_done(Capstone),
+            # final, done.
+            if kind == "final" and current_stone is not None:
+                yield _close(current_stone)
+                current_stone = None
+
             yield f"event: {kind}\ndata: {json.dumps(ev, default=str)}\n\n"
+
+        # Pipeline ended without a final (error / abort) — close any
+        # still-open Stone so the client doesn't render an unbounded
+        # parent row.
+        if current_stone is not None:
+            yield _close(current_stone)
         yield "event: done\ndata: {}\n\n"
 
     return StreamingResponse(event_stream(), media_type="text/event-stream",

web/static/agent.js

@@ -1237,6 +1237,32 @@ function ask(q) {
     if (step.step === "geocode" || step.step === "nta_resolve") setMapLoading(null);
   });
   es.addEventListener("step",  (e) => { pushTraceStep(JSON.parse(e.data)); });
+
+  // Stones envelope — `stone_start` and `stone_done` events bracket
+  // the contiguous step events of each Stone group. The current
+  // `<r-trace>` Svelte build doesn't yet render parent/child rows;
+  // we accumulate Stone markers in TRACE_BUF for the auditable report,
+  // and surface a lightweight badge on the trace component so users
+  // can see Cornerstone / Keystone / Touchstone / Lodestone / Capstone
+  // lighting up sequentially. The full collapsible parent-row UI
+  // lands once the trace component is rebuilt against this event
+  // vocabulary.
+  es.addEventListener("stone_start", (e) => {
+    const stone = JSON.parse(e.data);
+    TRACE_BUF.push({ _stone: "start", ...stone });
+    const trace = $("#trace");
+    if (trace && typeof trace.markStoneStart === "function") {
+      trace.markStoneStart(stone);
+    }
+  });
+  es.addEventListener("stone_done", (e) => {
+    const stone = JSON.parse(e.data);
+    TRACE_BUF.push({ _stone: "done", ...stone });
+    const trace = $("#trace");
+    if (trace && typeof trace.markStoneDone === "function") {
+      trace.markStoneDone(stone);
+    }
+  });
   let currentAttempt = 0;
   es.addEventListener("token", (e) => {
     const d = JSON.parse(e.data);