feat(miner-ui): useStreamingText hook + StreamingText renderer

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 (
+    <p className={className} data-status={status} aria-busy={status === "streaming"}>
+      {text}
+      {status === "streaming" && !reducedMotion ? (
+        <span aria-hidden="true" className="ml-0.5 inline-block animate-pulse">
+          ▍
+        </span>
+      ) : null}
+    </p>
+  );
+}

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<string>`,
+ * 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<string>;
+
+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<StreamingStatus>("idle");
+  const [error, setError] = useState<Error | null>(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 };
+}

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<void>((resolve) => (release = resolve));
+  const wake = () => {
+    const r = release;
+    release = null;
+    r?.();
+  };
+  async function* gen(): AsyncGenerator<string> {
+    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(<StreamingText source={null} />);
+    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(<StreamingText source={src.source} />);
+    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(<StreamingText source={src.source} />);
+    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");
+  });
+});

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<void>((resolve) => (release = resolve));
+  const wake = () => {
+    const r = release;
+    release = null;
+    r?.();
+  };
+  async function* gen(): AsyncGenerator<string> {
+    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");
+  });
+});