diff --git a/apps/loopover-miner-ui/src/components/streaming-text.tsx b/apps/loopover-miner-ui/src/components/streaming-text.tsx new file mode 100644 index 0000000000..f9398c60ae --- /dev/null +++ b/apps/loopover-miner-ui/src/components/streaming-text.tsx @@ -0,0 +1,41 @@ +import { useEffect, useState } from "react"; +import { useStreamingText, type ChunkSource } from "@/lib/use-streaming-text"; + +// prefers-reduced-motion detection via window.matchMedia + a `change` listener — the same technique +// packages/loopover-ui-kit/src/hooks/use-mobile.tsx uses. Kept internal (not exported) so this file only +// exports the component, satisfying react-refresh. This app has no `motion`/`framer-motion` dependency. +function usePrefersReducedMotion(): boolean { + const [reduced, setReduced] = useState(() => + typeof window !== "undefined" && typeof window.matchMedia === "function" + ? window.matchMedia("(prefers-reduced-motion: reduce)").matches + : false, + ); + useEffect(() => { + if (typeof window === "undefined" || typeof window.matchMedia !== "function") return; + const query = window.matchMedia("(prefers-reduced-motion: reduce)"); + const onChange = () => setReduced(query.matches); + query.addEventListener("change", onChange); + return () => query.removeEventListener("change", onChange); + }, []); + return reduced; +} + +/** + * Thin presentational renderer for {@link useStreamingText}: shows the progressively-accumulated text and, while + * streaming, a blinking caret. The reveal itself is never gated — only the caret animation is suppressed under + * prefers-reduced-motion, so reduced-motion users still see the full text arrive, just without the animation. + */ +export function StreamingText({ source, className }: { source: ChunkSource | null; className?: string }) { + const { text, status } = useStreamingText(source); + const reducedMotion = usePrefersReducedMotion(); + return ( +

+ {text} + {status === "streaming" && !reducedMotion ? ( + + ) : null} +

+ ); +} diff --git a/apps/loopover-miner-ui/src/lib/use-streaming-text.ts b/apps/loopover-miner-ui/src/lib/use-streaming-text.ts new file mode 100644 index 0000000000..81c8ddaf1f --- /dev/null +++ b/apps/loopover-miner-ui/src/lib/use-streaming-text.ts @@ -0,0 +1,80 @@ +import { useCallback, useEffect, useRef, useState } from "react"; + +/** + * A source of text chunks. A factory (invoked once per stream start) returning an `AsyncIterable`, + * so a caller can hand a fresh async generator or a `ReadableStream` wrapper each time — the hook never + * re-consumes an already-drained iterator. Exported by name so a later composer/message-list issue can type + * its streaming prop against it. + */ +export type ChunkSource = () => AsyncIterable; + +export type StreamingStatus = "idle" | "streaming" | "done" | "error" | "cancelled"; + +export interface StreamingTextState { + /** Text accumulated from all chunks consumed so far. */ + text: string; + status: StreamingStatus; + error: Error | null; + /** Stop consuming the current source; no later chunk from it reaches state. Idempotent, safe post-unmount. */ + cancel: () => void; +} + +/** + * Consume a chunked text source progressively (#6516): accumulate each chunk into `text` as it arrives and + * expose an idle/streaming/done/error/cancelled status. Mirrors `usePolledFetch`'s cancelled-flag discipline — + * a chunk resolving after a new source starts, after `cancel()`, or after unmount never touches state. This is + * an unwired primitive: it only ever consumes the source it's handed (a mock in tests, a real stream later). + */ +export function useStreamingText(source: ChunkSource | null): StreamingTextState { + const [text, setText] = useState(""); + const [status, setStatus] = useState("idle"); + const [error, setError] = useState(null); + // Points at the CURRENT effect's canceller so cancel() always targets the live stream, never a stale one. + const cancelRef = useRef<() => void>(() => {}); + + useEffect(() => { + // Per-effect flag (a fresh closure each run): the cleanup below flips it on a new source or unmount, so the + // previous run's worker stops and writes no more state. cancel() flips this same flag for an explicit stop. + let cancelled = false; + cancelRef.current = () => { + if (!cancelled) { + cancelled = true; + setStatus("cancelled"); + } + }; + + // ALL state writes live inside this async worker rather than the effect body — the reset + "streaming" + // transition, incremental accumulation, and terminal done/error transitions — so none is a synchronous + // setState-in-effect (react-hooks/set-state-in-effect). Each is guarded by `cancelled` so a write never + // lands after a new source starts, after cancel(), or after unmount. + void (async () => { + if (cancelled) return; + setText(""); + setError(null); + if (!source) { + setStatus("idle"); + return; + } + setStatus("streaming"); + try { + for await (const chunk of source()) { + if (cancelled) return; + setText((prev) => prev + chunk); + } + if (!cancelled) setStatus("done"); + } catch (err) { + if (!cancelled) { + setError(err instanceof Error ? err : new Error(String(err))); + setStatus("error"); + } + } + })(); + + return () => { + cancelled = true; + }; + }, [source]); + + const cancel = useCallback(() => cancelRef.current(), []); + return { text, status, error, cancel }; +} diff --git a/apps/loopover-miner-ui/src/streaming-text.test.tsx b/apps/loopover-miner-ui/src/streaming-text.test.tsx new file mode 100644 index 0000000000..ad84beb7d4 --- /dev/null +++ b/apps/loopover-miner-ui/src/streaming-text.test.tsx @@ -0,0 +1,78 @@ +import { act, render, screen, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { StreamingText } from "./components/streaming-text"; +import type { ChunkSource } from "./lib/use-streaming-text"; + +afterEach(() => vi.unstubAllGlobals()); + +function mockReducedMotion(reduced: boolean) { + vi.stubGlobal("matchMedia", (query: string) => ({ + matches: reduced && query.includes("reduce"), + media: query, + onchange: null, + addEventListener: () => {}, + removeEventListener: () => {}, + addListener: () => {}, + removeListener: () => {}, + dispatchEvent: () => false, + })); +} + +/** Caller-driven chunk source so a test can hold the component in its "streaming" state to assert the caret. */ +function deferredSource() { + const queued: string[] = []; + let release: (() => void) | null = null; + let finished = false; + const gate = () => new Promise((resolve) => (release = resolve)); + const wake = () => { + const r = release; + release = null; + r?.(); + }; + async function* gen(): AsyncGenerator { + let i = 0; + for (;;) { + while (i < queued.length) yield queued[i++]!; + if (finished) return; + await gate(); + } + } + return { + source: (() => gen()) as ChunkSource, + push: async (chunk: string) => act(async () => (queued.push(chunk), wake())), + finish: async () => act(async () => ((finished = true), wake())), + }; +} + +const caret = () => document.querySelector("span[aria-hidden='true']"); + +describe("StreamingText (#6516)", () => { + it("renders an idle paragraph with no text when given no source", () => { + mockReducedMotion(false); + const { container } = render(); + expect(container.querySelector("p")?.getAttribute("data-status")).toBe("idle"); + expect(container.textContent).toBe(""); + }); + + it("reveals accumulated text and shows an animated caret while streaming (full motion)", async () => { + mockReducedMotion(false); + const src = deferredSource(); + render(); + await src.push("typing…"); + await waitFor(() => expect(screen.getByText(/typing…/)).toBeTruthy()); + expect(caret()).not.toBeNull(); // still streaming → caret present under full motion + }); + + it("suppresses the caret under prefers-reduced-motion but still reaches the full text and done", async () => { + mockReducedMotion(true); + const src = deferredSource(); + render(); + await src.push("no caret here"); + await waitFor(() => expect(screen.getByText(/no caret here/)).toBeTruthy()); + expect(caret()).toBeNull(); // reduced motion → no animated caret even mid-stream + + await src.finish(); + await waitFor(() => expect(document.querySelector("p")?.getAttribute("data-status")).toBe("done")); + expect(document.querySelector("p")?.textContent).toContain("no caret here"); + }); +}); diff --git a/apps/loopover-miner-ui/src/use-streaming-text.test.ts b/apps/loopover-miner-ui/src/use-streaming-text.test.ts new file mode 100644 index 0000000000..a0554b15ee --- /dev/null +++ b/apps/loopover-miner-ui/src/use-streaming-text.test.ts @@ -0,0 +1,117 @@ +import { act, renderHook, waitFor } from "@testing-library/react"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { useStreamingText, type ChunkSource } from "./lib/use-streaming-text"; + +afterEach(() => vi.restoreAllMocks()); + +/** + * A chunk source whose delivery the test drives explicitly: the generator yields any queued chunks, then + * awaits a gate; `push`/`fail`/`finish` enqueue the next event and release the gate. This makes intermediate + * accumulation and the cancel/error transitions deterministic, in the spirit of use-polled-fetch.test.ts's + * fake-timer control. + */ +function deferredSource() { + const queued: string[] = []; + let release: (() => void) | null = null; + let finished = false; + let failure: Error | null = null; + const gate = () => new Promise((resolve) => (release = resolve)); + const wake = () => { + const r = release; + release = null; + r?.(); + }; + async function* gen(): AsyncGenerator { + let i = 0; + for (;;) { + while (i < queued.length) yield queued[i++]!; + if (failure) throw failure; + if (finished) return; + await gate(); + } + } + return { + source: (() => gen()) as ChunkSource, + push: async (chunk: string) => act(async () => (queued.push(chunk), wake())), + fail: async (err: Error) => act(async () => ((failure = err), wake())), + finish: async () => act(async () => ((finished = true), wake())), + }; +} + +describe("useStreamingText (#6516)", () => { + it("starts idle when given no source", () => { + const { result } = renderHook(() => useStreamingText(null)); + expect(result.current).toMatchObject({ text: "", status: "idle", error: null }); + }); + + it("accumulates chunks incrementally across renders, then reaches done", async () => { + const src = deferredSource(); + const { result } = renderHook(() => useStreamingText(src.source)); + await waitFor(() => expect(result.current.status).toBe("streaming")); + + await src.push("Hel"); + await waitFor(() => expect(result.current.text).toBe("Hel")); + await src.push("lo wor"); + await waitFor(() => expect(result.current.text).toBe("Hello wor")); + await src.push("ld"); + await waitFor(() => expect(result.current.text).toBe("Hello world")); + + await src.finish(); + await waitFor(() => expect(result.current.status).toBe("done")); + expect(result.current.text).toBe("Hello world"); + }); + + it("cancel() stops the stream and no later chunk reaches state", async () => { + const src = deferredSource(); + const { result } = renderHook(() => useStreamingText(src.source)); + await src.push("first"); + await waitFor(() => expect(result.current.text).toBe("first")); + + act(() => result.current.cancel()); + await waitFor(() => expect(result.current.status).toBe("cancelled")); + + await src.push("late"); // arrives after cancel — must be ignored + expect(result.current.text).toBe("first"); + expect(result.current.status).toBe("cancelled"); + }); + + it("starting a new source stops the previous one; its late chunk never reaches state", async () => { + const first = deferredSource(); + const { result, rerender } = renderHook(({ s }: { s: ChunkSource }) => useStreamingText(s), { + initialProps: { s: first.source }, + }); + await first.push("old"); + await waitFor(() => expect(result.current.text).toBe("old")); + + const second = deferredSource(); + rerender({ s: second.source }); // swap sources mid-stream + await second.push("new"); + await waitFor(() => expect(result.current.text).toBe("new")); + + await first.push("STALE"); // a late chunk from the abandoned first source + expect(result.current.text).toBe("new"); + }); + + it("does not update state after unmount, even if a chunk resolves late", async () => { + const src = deferredSource(); + const { result, unmount } = renderHook(() => useStreamingText(src.source)); + await src.push("kept"); + await waitFor(() => expect(result.current.text).toBe("kept")); + + unmount(); + await expect(src.push("after-unmount")).resolves.not.toThrow(); // no throw, no state write + expect(result.current.text).toBe("kept"); + }); + + it("surfaces a mid-stream error through status/error, not as an unhandled rejection", async () => { + const src = deferredSource(); + const { result } = renderHook(() => useStreamingText(src.source)); + await src.push("partial"); + await waitFor(() => expect(result.current.text).toBe("partial")); + + await src.fail(new Error("stream boom")); + await waitFor(() => expect(result.current.status).toBe("error")); + expect(result.current.error?.message).toBe("stream boom"); + expect(result.current.text).toBe("partial"); + }); +});