feat(console): forward Last-Event-ID through SSE proxy

The console SSE proxy (`_proxy_sse`) is the inbound SSE path for
multi-node deployments — every browser EventSource that targets a
per-node route traverses it.  Today's proxy strips client request
headers (only `Accept`, `Cache-Control`, and the re-minted auth
token make it upstream), so the per-ws / global SSE handlers'
`Last-Event-ID` resume (PR-D commit 1) never sees the header in
the multi-node shape — every reconnect would be a fresh connect
and silently drop events from the disconnect window.

Builds the upstream headers dict conditionally: copy `Last-Event-ID`
from the incoming request when present, omit otherwise (no
fabricated value on fresh connects).  Starlette's header dict is
case-insensitive so the `request.headers.get("last-event-id")`
lookup catches both the spec-recommended capitalization and any
intermediary normalisation.

The query-param fallback (`?last_event_id=N`) needs no proxy
change — `request.url.query` is already forwarded verbatim at the
top of the function.

Tests in `tests/test_service_auth_boundary.py::TestProxySseLastEventIdForwarding`:
- Positive: browser header → upstream header (value preserved).
- Negative: browser sends nothing → upstream gets nothing (no
  fabricated value).

tests/test_service_auth_boundary.py

@@ -440,6 +440,108 @@ class TestProxySseNon200LogLevel:
         assert "\n" not in matches[0].getMessage().split("body=", 1)[-1]
 
 
+# ---------------------------------------------------------------------------
+# _proxy_sse — Last-Event-ID forwarding (PR-D reconnect-with-replay)
+# ---------------------------------------------------------------------------
+
+
+class TestProxySseLastEventIdForwarding:
+    """The console SSE proxy is the inbound SSE path for multi-node
+    deployments — every browser EventSource traverses it.  Without
+    forwarding ``Last-Event-ID``, the per-ws / global SSE handlers on
+    the node would treat every reconnect as a fresh connect and silently
+    drop events emitted during the disconnect window.  PR-D's whole
+    reconnect-with-replay foundation depends on these tests passing."""
+
+    @pytest.mark.anyio
+    async def test_forwards_last_event_id_header_to_upstream(self):
+        """Browser sends ``Last-Event-ID``; upstream node must receive it."""
+        from starlette.requests import Request
+
+        from turnstone.console.server import _proxy_sse
+
+        captured_headers: dict[str, str] = {}
+
+        def handler(req: httpx.Request) -> httpx.Response:
+            # httpx headers are case-insensitive; capture lowercased.
+            captured_headers.update({k.lower(): v for k, v in req.headers.items()})
+            return httpx.Response(
+                200, text="data: {}\n\n", headers={"content-type": "text/event-stream"}
+            )
+
+        sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+        proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+        scope = {
+            "type": "http",
+            "method": "GET",
+            "path": "/node/n/api/workstreams/ws-1/events",
+            "headers": [(b"last-event-id", b"42")],
+            "query_string": b"",
+            "app": MagicMock(
+                state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client)
+            ),
+        }
+
+        async def _receive():
+            return {"type": "http.request", "body": b""}
+
+        request = Request(scope, receive=_receive)
+        response = await _proxy_sse(
+            request, "http://node-1:8001", "workstreams/ws-1/events", api_prefix="api"
+        )
+        # Drain so the upstream call actually fires.
+        async for _ in response.body_iterator:  # type: ignore[attr-defined]
+            pass
+
+        assert captured_headers.get("last-event-id") == "42", (
+            f"Last-Event-ID not forwarded to upstream; got headers={captured_headers!r}"
+        )
+
+    @pytest.mark.anyio
+    async def test_omits_last_event_id_when_client_did_not_send_one(self):
+        """Fresh connect (no header on the browser side) → no header
+        added on the upstream side either.  Guards against
+        accidentally injecting a stale or fabricated value."""
+        from starlette.requests import Request
+
+        from turnstone.console.server import _proxy_sse
+
+        captured_headers: dict[str, str] = {}
+
+        def handler(req: httpx.Request) -> httpx.Response:
+            captured_headers.update({k.lower(): v for k, v in req.headers.items()})
+            return httpx.Response(
+                200, text="data: {}\n\n", headers={"content-type": "text/event-stream"}
+            )
+
+        sse_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+        proxy_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+        scope = {
+            "type": "http",
+            "method": "GET",
+            "path": "/node/n/api/workstreams/ws-1/events",
+            "headers": [],
+            "query_string": b"",
+            "app": MagicMock(
+                state=SimpleNamespace(proxy_sse_client=sse_client, proxy_client=proxy_client)
+            ),
+        }
+
+        async def _receive():
+            return {"type": "http.request", "body": b""}
+
+        request = Request(scope, receive=_receive)
+        response = await _proxy_sse(
+            request, "http://node-1:8001", "workstreams/ws-1/events", api_prefix="api"
+        )
+        async for _ in response.body_iterator:  # type: ignore[attr-defined]
+            pass
+
+        assert "last-event-id" not in captured_headers, (
+            f"upstream got an unexpected Last-Event-ID; headers={captured_headers!r}"
+        )
+
+
 # ---------------------------------------------------------------------------
 # Gated cluster_events_sse — 503 on scope error (0a)
 # ---------------------------------------------------------------------------

turnstone/console/server.py

@@ -2866,12 +2866,31 @@ async def _proxy_sse(
     else:
         sse_auth = _proxy_auth_headers(request)
 
+    # Forward ``Last-Event-ID`` from the browser to the upstream node so
+    # the per-ws / global SSE handlers can serve the reconnect-with-replay
+    # buffer slice.  Without this, multi-node deployments lose replay
+    # entirely (the proxy is the only inbound SSE path in that shape);
+    # the per-ws handler would treat every reconnect as a fresh connect
+    # and silently drop events emitted during the disconnect window.
+    # The query-param fallback (``?last_event_id=N``) is already
+    # forwarded via the existing ``request.url.query`` propagation at
+    # the top of this function — only the header needs an explicit
+    # carry-over.  Header lookups in Starlette are case-insensitive.
+    upstream_headers: dict[str, str] = {
+        **sse_auth,
+        "Accept": "text/event-stream",
+        "Cache-Control": "no-store",
+    }
+    last_event_id_hdr = request.headers.get("last-event-id")
+    if last_event_id_hdr is not None:
+        upstream_headers["Last-Event-ID"] = last_event_id_hdr
+
     async def raw_stream() -> AsyncGenerator[bytes, None]:
         try:
             async with sse_client.stream(
                 "GET",
                 target,
-                headers={**sse_auth, "Accept": "text/event-stream", "Cache-Control": "no-store"},
+                headers=upstream_headers,
                 timeout=httpx.Timeout(connect=10, read=None, write=5, pool=None),
             ) as response:
                 if response.status_code != 200: