feat(miner-ui): wire the chat rail to the read-only streaming backend (#6518)

apps/loopover-miner-ui/src/chat-conversation.test.tsx

@@ -0,0 +1,78 @@
+import { fireEvent, render, screen, waitFor } from "@testing-library/react";
+import { describe, expect, it } from "vitest";
+
+import { ChatConversation } from "./components/chat/conversation";
+import type { ChatWireMessage } from "./lib/chat-stream";
+
+const sendButton = () => screen.getByRole("button", { name: "Send" }) as HTMLButtonElement;
+
+function ask(question: string) {
+  fireEvent.change(screen.getByRole("textbox"), { target: { value: question } });
+  fireEvent.click(screen.getByRole("button", { name: "Send" }));
+}
+
+/** A promise plus its resolver, for gating a stream open across an assertion. */
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((r) => (resolve = r));
+  return { promise, resolve };
+}
+
+describe("ChatConversation (#6518)", () => {
+  it("renders the empty conversation state with an enabled composer before any question", () => {
+    render(
+      <ChatConversation
+        streamChatImpl={async function* () {
+          /* no messages */
+        }}
+      />,
+    );
+    expect(screen.getByText(/No messages yet/i)).toBeTruthy();
+    expect(sendButton().disabled).toBe(false);
+  });
+
+  it("sends the composed question to the backend as wire-shaped history", async () => {
+    const seen: ChatWireMessage[][] = [];
+    const streamChatImpl = async function* (messages: ChatWireMessage[]) {
+      seen.push(messages);
+      yield "ok";
+    };
+    render(<ChatConversation streamChatImpl={streamChatImpl} />);
+    ask("what is stuck?");
+    await waitFor(() => expect(screen.getByText("ok")).toBeTruthy());
+    expect(seen[0]).toEqual([{ role: "user", content: "what is stuck?" }]);
+  });
+
+  it("disables the composer while a response streams, commits the answer, and re-enables it", async () => {
+    const gate = deferred();
+    const streamChatImpl = async function* (_messages: ChatWireMessage[]) {
+      yield "Hel";
+      await gate.promise;
+      yield "lo";
+    };
+    render(<ChatConversation streamChatImpl={streamChatImpl} />);
+    ask("hi");
+
+    // The question shows immediately and the composer is locked for the whole in-flight window.
+    await waitFor(() => expect(sendButton().disabled).toBe(true));
+    expect(screen.getByText("hi")).toBeTruthy();
+
+    gate.resolve();
+
+    // On completion the streamed answer is committed into the list and the composer re-enables.
+    await waitFor(() => expect(sendButton().disabled).toBe(false));
+    expect(screen.getByText("Hello")).toBeTruthy();
+  });
+
+  it("surfaces a backend failure through the message-list error state and re-enables the composer", async () => {
+    const streamChatImpl = async function* (_messages: ChatWireMessage[]): AsyncGenerator<string> {
+      yield* []; // yields nothing, then fails — models a backend/stream error mid-request
+      throw new Error("connection refused");
+    };
+    render(<ChatConversation streamChatImpl={streamChatImpl} />);
+    ask("hi");
+
+    await waitFor(() => expect(screen.getByText(/Couldn't load the conversation/i)).toBeTruthy());
+    expect(sendButton().disabled).toBe(false);
+  });
+});

apps/loopover-miner-ui/src/chat-stream.test.ts

@@ -0,0 +1,90 @@
+import { describe, expect, it } from "vitest";
+
+import { CHAT_API_PATH, streamChat, type ChatWireMessage } from "./lib/chat-stream";
+
+/** Build a fake `text/event-stream` Response whose body streams the given raw SSE chunks (each may hold part of,
+ *  one, or several `data:` frames — the point is to exercise the client's frame reassembly). */
+function sseResponse(chunks: string[], status = 200): Response {
+  const body = new ReadableStream<Uint8Array>({
+    start(controller) {
+      const encoder = new TextEncoder();
+      for (const chunk of chunks) controller.enqueue(encoder.encode(chunk));
+      controller.close();
+    },
+  });
+  return new Response(body, { status, headers: { "Content-Type": "text/event-stream" } });
+}
+
+async function collect(stream: AsyncIterable<string>): Promise<string[]> {
+  const out: string[] = [];
+  for await (const delta of stream) out.push(delta);
+  return out;
+}
+
+const HI: ChatWireMessage[] = [{ role: "user", content: "hi" }];
+
+describe("streamChat (#6518)", () => {
+  it("yields each text delta, stops at the done event, and POSTs the messages to /api/chat", async () => {
+    let url: string | undefined;
+    let init: RequestInit | undefined;
+    const chunks = await collect(
+      streamChat(HI, async (input, requestInit) => {
+        url = input;
+        init = requestInit;
+        return sseResponse([
+          'data: {"type":"text","text":"Hel"}\n\n',
+          'data: {"type":"text","text":"lo"}\n\n',
+          'data: {"type":"done"}\n\n',
+          'data: {"type":"text","text":"AFTER-DONE"}\n\n', // never reached: done ends the stream
+        ]);
+      }),
+    );
+    expect(chunks).toEqual(["Hel", "lo"]);
+    expect(url).toBe(CHAT_API_PATH);
+    expect(init?.method).toBe("POST");
+    expect(JSON.parse(String(init?.body))).toEqual({ messages: HI });
+  });
+
+  it("consumes and skips grounding tool events, yielding only display text", async () => {
+    const chunks = await collect(
+      streamChat(HI, async () =>
+        sseResponse([
+          'data: {"type":"tool_call","tool":"loopover_miner_run_state","input":{}}\n\n',
+          'data: {"type":"text","text":"A"}\n\n',
+          'data: {"type":"tool_result","tool":"loopover_miner_run_state","output":{}}\n\n',
+          'data: {"type":"text","text":"B"}\n\n',
+          'data: {"type":"done"}\n\n',
+        ]),
+      ),
+    );
+    expect(chunks).toEqual(["A", "B"]);
+  });
+
+  it("reassembles a single event split across read boundaries", async () => {
+    const chunks = await collect(
+      streamChat(HI, async () =>
+        sseResponse(['data: {"type":"te', 'xt","text":"split"}\n\n', 'data: {"type":"done"}\n\n']),
+      ),
+    );
+    expect(chunks).toEqual(["split"]);
+  });
+
+  it("throws when the stream emits an error event", async () => {
+    await expect(
+      collect(
+        streamChat(HI, async () =>
+          sseResponse([
+            'data: {"type":"text","text":"partial"}\n\n',
+            'data: {"type":"error","code":"boom","message":"grounding failed"}\n\n',
+          ]),
+        ),
+      ),
+    ).rejects.toThrow(/grounding failed/);
+  });
+
+  it("throws before any delta on a non-2xx (validation) response", async () => {
+    await expect(
+      collect(streamChat([], async () => new Response(JSON.stringify({ error: "bad messages" }), { status: 400 }))),
+    ).rejects.toThrow(/400/);
+  });
+});

apps/loopover-miner-ui/src/components/chat-rail.tsx

@@ -1,25 +1,24 @@
 // Persistent chat-rail shell (#6513). A pure structural shell mounted once in __root.tsx so it survives
 // client-side route navigation: on wide viewports it docks as a ~380px panel beside the routed content; below
 // the ui-kit `useIsMobile` breakpoint it collapses to the same `Sheet`-based slide-over `sidebar.tsx` uses for
-// its own mobile mode (rather than a second, bespoke mobile-collapse mechanism). This ships with static
-// placeholder content only — no composer, message list, streaming, or backend call; those layer on later.
+// its own mobile mode (rather than a second, bespoke mobile-collapse mechanism). The rail's content slot is
+// filled by the `ChatConversation` integration (#6518) — composer + message list + streaming renderer wired to
+// the read-only `POST /api/chat` backend — rendered by both the docked panel and the mobile sheet.
 import * as React from "react";
 
 import { Button } from "@loopover/ui-kit/components/button";
 import { Sheet, SheetContent, SheetDescription, SheetHeader, SheetTitle } from "@loopover/ui-kit/components/sheet";
 import { useIsMobile } from "@loopover/ui-kit/hooks/use-mobile";
 
+import { ChatConversation } from "@/components/chat/conversation";
+
 const RAIL_WIDTH_PX = 380;
 const RAIL_PANEL_ID = "chat-rail-panel";
 
-/** The rail's inner content. Static placeholder for this shell issue — the real composer/message-list land later. */
+/** The rail's inner content slot, filled by the read-only chat conversation integration (#6518). Rendered by
+ *  both the mobile slide-over sheet and the desktop docked panel so a single wiring covers both presentations. */
 function RailBody() {
-  return (
-    <div className="flex h-full flex-col gap-2 p-4">
-      <p className="font-mono text-token-xs uppercase tracking-[0.2em] text-primary">Chat</p>
-      <p className="text-token-sm text-muted-foreground">Ask about this miner&rsquo;s local state. Coming soon.</p>
-    </div>
-  );
+  return <ChatConversation />;
 }
 
 export interface ChatRailProps {

apps/loopover-miner-ui/src/components/chat/conversation.tsx

@@ -0,0 +1,107 @@
+import { useCallback, useRef, useState } from "react";
+
+import { Avatar, AvatarFallback } from "@loopover/ui-kit/components/avatar";
+
+import { ChatComposer } from "@/components/chat-composer";
+import { StreamingText } from "@/components/streaming-text";
+import { MessageList } from "@/components/chat/message-list";
+import type { ChatMessage } from "@/components/chat/fixtures";
+import type { ChunkSource } from "@/lib/use-streaming-text";
+import { streamChat, type ChatWireMessage } from "@/lib/chat-stream";
+
+// The chat-rail's content integration (#6518): the first point the persistent rail (#6513) holds a live
+// conversation. Pure wiring — it composes the standalone composer (#6514), message list (#6515), and streaming
+// renderer (#6516) around the read-only streaming backend (#6517), and owns nothing but the conversation state.
+// Strictly ask-a-question / read-only: the only network call it can make is `streamChat` → `POST /api/chat`; it
+// never touches an action endpoint (portfolio release/requeue, governor pause/resume) — that surface is a
+// separate, later, flag-gated issue.
+
+const ASSISTANT_NAME = "LoopOver";
+
+/** Injectable so tests can drive the stream deterministically; defaults to the real `POST /api/chat` bridge. */
+export type StreamChatFn = (messages: ChatWireMessage[]) => AsyncIterable<string>;
+
+export function ChatConversation({ streamChatImpl = streamChat }: { streamChatImpl?: StreamChatFn } = {}) {
+  const [messages, setMessages] = useState<ChatMessage[]>([]);
+  const [activeSource, setActiveSource] = useState<ChunkSource | null>(null);
+  const [streaming, setStreaming] = useState(false);
+  const [errored, setErrored] = useState(false);
+  const idCounter = useRef(0);
+  const nextId = () => `m${(idCounter.current += 1)}`;
+
+  const handleSubmit = useCallback(
+    (text: string) => {
+      const userMessage: ChatMessage = {
+        id: nextId(),
+        role: "user",
+        content: text,
+        timestamp: new Date().toISOString(),
+      };
+      // What the backend grounds against: the prior user/assistant turns plus this question, in wire shape.
+      const history: ChatWireMessage[] = [...messages, userMessage]
+        .filter((message): message is ChatMessage & { role: "user" | "assistant" } => message.role !== "system")
+        .map((message) => ({ role: message.role, content: message.content }));
+
+      setMessages((prev) => [...prev, userMessage]);
+      setErrored(false);
+      setStreaming(true);
+
+      // This source both feeds the live StreamingText render AND, on natural completion, commits the finished
+      // answer into the message list. Every state write below runs in the generator's async continuation (driven
+      // by useStreamingText inside StreamingText), never synchronously in an effect body — so it stays clear of
+      // react-hooks/set-state-in-effect. The composer is disabled for the whole in-flight window, so a second
+      // request can't start before this one resolves and clears `streaming`.
+      const source: ChunkSource = () =>
+        (async function* () {
+          let answer = "";
+          try {
+            for await (const delta of streamChatImpl(history)) {
+              answer += delta;
+              yield delta;
+            }
+            setMessages((prev) => [
+              ...prev,
+              {
+                id: nextId(),
+                role: "assistant",
+                content: answer,
+                timestamp: new Date().toISOString(),
+                authorName: ASSISTANT_NAME,
+              },
+            ]);
+          } catch {
+            setErrored(true);
+          } finally {
+            setStreaming(false);
+            setActiveSource(null);
+          }
+        })();
+
+      // `source` is itself a function, so it must be stored via an updater — a bare `setActiveSource(source)`
+      // would be read as a functional update and *call* it instead of storing it.
+      setActiveSource(() => source);
+    },
+    [messages, streamChatImpl],
+  );
+
+  return (
+    <div className="flex h-full flex-col gap-2 p-4">
+      <p className="font-mono text-token-xs uppercase tracking-[0.2em] text-primary">Chat</p>
+      <div className="min-h-0 flex-1 overflow-hidden">
+        <MessageList messages={messages} isError={errored} />
+        {streaming && activeSource ? (
+          <div className="flex gap-3 px-3 pt-4" data-testid="chat-streaming-response">
+            <Avatar className="size-8 shrink-0">
+              <AvatarFallback>{ASSISTANT_NAME.slice(0, 2).toUpperCase()}</AvatarFallback>
+            </Avatar>
+            <StreamingText
+              source={activeSource}
+              className="min-w-0 whitespace-pre-wrap break-words rounded-token-sm bg-muted px-3 py-2 text-token-sm text-foreground"
+            />
+          </div>
+        ) : null}
+      </div>
+      <ChatComposer onSubmit={handleSubmit} disabled={streaming} />
+    </div>
+  );
+}

apps/loopover-miner-ui/src/lib/chat-stream.ts

@@ -0,0 +1,79 @@
+// Client bridge to the read-only streaming chat endpoint (#6518, wiring the rail to the #6517 backend).
+// `POST /api/chat` answers as `text/event-stream` — one `data: <json>\n\n` frame per event, consumed here via
+// fetch() + ReadableStream (not the native EventSource, which can't send the POST body the endpoint needs).
+// This yields only the human-readable `text` deltas as an async string stream, so it slots straight in as a
+// `ChunkSource` for the shared `useStreamingText`/`StreamingText` renderer; grounding `tool_call`/`tool_result`
+// frames are consumed and skipped (they carry no display text), `error` frames reject, and `done` ends the
+// stream. A validation failure comes back as a plain non-streamed 4xx JSON body, surfaced here as a thrown error
+// before any token is yielded.
+
+/** The wire event union the endpoint emits (mirrored from vite-chat-api.ts's `ChatSseEvent`; the app
+ *  deliberately doesn't import the server plugin module just for its type). */
+export type ChatStreamEvent =
+  | { type: "text"; text: string }
+  | { type: "tool_call"; tool: string; input: Record<string, unknown> }
+  | { type: "tool_result"; tool: string; output: unknown }
+  | { type: "error"; code: string; message: string }
+  | { type: "done" };
+
+/** The message shape the endpoint grounds against — only the two conversational roles cross the wire. */
+export type ChatWireMessage = { role: "user" | "assistant"; content: string };
+
+export const CHAT_API_PATH = "/api/chat";
+
+type FetchImpl = (input: string, init: RequestInit) => Promise<Response>;
+
+/** Parse one accumulated SSE frame (its `data:` line) into a typed event, or null when it carries no data. */
+function parseFrame(frame: string): ChatStreamEvent | null {
+  const dataLine = frame.split("\n").find((line) => line.startsWith("data:"));
+  if (!dataLine) return null;
+  const payload = dataLine.slice("data:".length).trim();
+  if (!payload) return null;
+  return JSON.parse(payload) as ChatStreamEvent;
+}
+
+/**
+ * Open the chat stream for `messages` and yield each `text` delta as it arrives. Throws before the first yield on
+ * a non-2xx / non-streamed response, and mid-stream on an `error` frame, so the caller's stream renderer lands in
+ * its error state. Injectable `fetchImpl` keeps it unit-testable against a fake ReadableStream.
+ */
+export async function* streamChat(
+  messages: ChatWireMessage[],
+  fetchImpl: FetchImpl = (input, init) => fetch(input, init),
+): AsyncGenerator<string> {
+  const response = await fetchImpl(CHAT_API_PATH, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ messages }),
+  });
+  if (!response.ok || !response.body) {
+    throw new Error(`chat backend responded ${response.status}`);
+  }
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      // SSE frames are delimited by a blank line; drain every complete frame the buffer now holds.
+      let boundary = buffer.indexOf("\n\n");
+      while (boundary !== -1) {
+        const frame = buffer.slice(0, boundary);
+        buffer = buffer.slice(boundary + 2);
+        const event = parseFrame(frame);
+        if (event) {
+          if (event.type === "text") yield event.text;
+          else if (event.type === "error") throw new Error(event.message || event.code || "chat stream error");
+          else if (event.type === "done") return;
+          // tool_call / tool_result carry grounding, not display text — consume and skip.
+        }
+        boundary = buffer.indexOf("\n\n");
+      }
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}