diff --git a/apps/loopover-miner-ui/src/chat-conversation.test.tsx b/apps/loopover-miner-ui/src/chat-conversation.test.tsx new file mode 100644 index 0000000000..af7f9b5a91 --- /dev/null +++ b/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((r) => (resolve = r)); + return { promise, resolve }; +} + +describe("ChatConversation (#6518)", () => { + it("renders the empty conversation state with an enabled composer before any question", () => { + render( + , + ); + 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(); + 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(); + 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 { + yield* []; // yields nothing, then fails — models a backend/stream error mid-request + throw new Error("connection refused"); + }; + render(); + ask("hi"); + + await waitFor(() => expect(screen.getByText(/Couldn't load the conversation/i)).toBeTruthy()); + expect(sendButton().disabled).toBe(false); + }); +}); diff --git a/apps/loopover-miner-ui/src/chat-stream.test.ts b/apps/loopover-miner-ui/src/chat-stream.test.ts new file mode 100644 index 0000000000..2887fffaf7 --- /dev/null +++ b/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({ + 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): Promise { + 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/); + }); +}); diff --git a/apps/loopover-miner-ui/src/components/chat-rail.tsx b/apps/loopover-miner-ui/src/components/chat-rail.tsx index 52adee8e22..53e3b9494e 100644 --- a/apps/loopover-miner-ui/src/components/chat-rail.tsx +++ b/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 ( -
-

Chat

-

Ask about this miner’s local state. Coming soon.

-
- ); + return ; } export interface ChatRailProps { diff --git a/apps/loopover-miner-ui/src/components/chat/conversation.tsx b/apps/loopover-miner-ui/src/components/chat/conversation.tsx new file mode 100644 index 0000000000..e55dcba33f --- /dev/null +++ b/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; + +export function ChatConversation({ streamChatImpl = streamChat }: { streamChatImpl?: StreamChatFn } = {}) { + const [messages, setMessages] = useState([]); + const [activeSource, setActiveSource] = useState(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 ( +
+

Chat

+
+ + {streaming && activeSource ? ( +
+ + {ASSISTANT_NAME.slice(0, 2).toUpperCase()} + + +
+ ) : null} +
+ +
+ ); +} diff --git a/apps/loopover-miner-ui/src/lib/chat-stream.ts b/apps/loopover-miner-ui/src/lib/chat-stream.ts new file mode 100644 index 0000000000..0da1d7d767 --- /dev/null +++ b/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: \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 } + | { 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; + +/** 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 { + 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(); + } +}