/** * @file useRunStream.test.tsx * @description Covers `useRunStream`, the hook that owns the Run page's live * envelope state: it subscribes to the event bus and folds `run_stream` * envelopes into an array, forwards `run_status` / `run_input_ack` for the * subscribed run id to the caller's callbacks, fires the id-agnostic * `onAnyStatus` for every `run_status`, and disposes its subscription on * unmount. `eventBus` is exercised for real (it is a plain in-memory pub/sub) * with only its `subscribe` spied on, so the disposer assertion pins the real * lifecycle rather than a mock's. * * @author Nguyễn Ngọc Trí Vĩ */ import { act, renderHook } from "@testing-library/react"; import { afterEach, describe, expect, it, vi } from "vitest"; import { eventBus } from "../../lib/eventBus"; import type { WSMessage } from "../../lib/types"; import { useRunStream, type Envelope } from "../useRunStream"; /** `run_stream` frame carrying one envelope for `id`. */ function streamMsg(id: string, envelope: unknown): WSMessage { return { type: "run_stream", data: { id, envelope } } as WSMessage; } function statusMsg(id: string, status: string): WSMessage { return { type: "run_status", data: { id, status, at: 1 } } as WSMessage; } const noopOpts = { onStatus: () => {}, onInputAck: () => {}, onAnyStatus: () => {} }; afterEach(() => { vi.restoreAllMocks(); }); describe("useRunStream", () => { it("merges envelopes for the subscribed run id in arrival order", () => { const { result } = renderHook(() => useRunStream("run-1", noopOpts)); act(() => { eventBus.publish(streamMsg("run-1", { type: "system", subtype: "init" })); eventBus.publish(streamMsg("run-1", { type: "result", subtype: "success" })); }); expect(result.current.envelopes.map((e) => (e as { type: string }).type)).toEqual([ "system", "result", ]); }); it("ignores an envelope for a different run id", () => { const { result } = renderHook(() => useRunStream("run-1", noopOpts)); act(() => { eventBus.publish(streamMsg("run-2", { type: "result" })); }); expect(result.current.envelopes).toEqual([]); }); it("updates a streaming assistant envelope in place instead of appending", () => { const { result } = renderHook(() => useRunStream("run-1", noopOpts)); act(() => { // message_start seeds the kept envelope + a streaming placeholder. eventBus.publish( streamMsg("run-1", { type: "stream_event", event: { type: "message_start", message: { id: "m1" } }, }) ); }); expect(result.current.envelopes).toHaveLength(2); act(() => { eventBus.publish( streamMsg("run-1", { type: "stream_event", event: { type: "content_block_start", index: 0, message: { id: "m1" }, content_block: { type: "text", text: "" }, }, }) ); eventBus.publish( streamMsg("run-1", { type: "stream_event", event: { type: "content_block_delta", index: 0, message: { id: "m1" }, delta: { type: "text_delta", text: "hi" }, }, }) ); }); // Still 2 envelopes: the deltas mutated the placeholder, they did not append. expect(result.current.envelopes).toHaveLength(2); const placeholder = result.current.envelopes[1] as { message: { content: { text?: string }[]; _streaming?: boolean }; }; expect(placeholder.message.content[0]?.text).toBe("hi"); expect(placeholder.message._streaming).toBe(true); }); it("invokes onStatus only for the subscribed run id, onAnyStatus for every run_status", () => { const onStatus = vi.fn(); const onAnyStatus = vi.fn(); renderHook(() => useRunStream("run-1", { ...noopOpts, onStatus, onAnyStatus })); act(() => { eventBus.publish(statusMsg("run-1", "completed")); eventBus.publish(statusMsg("run-2", "completed")); }); expect(onStatus).toHaveBeenCalledTimes(1); expect(onStatus.mock.calls[0]?.[0]).toMatchObject({ id: "run-1", status: "completed" }); expect(onAnyStatus).toHaveBeenCalledTimes(2); }); it("invokes onInputAck only for the subscribed run id", () => { const onInputAck = vi.fn(); renderHook(() => useRunStream("run-1", { ...noopOpts, onInputAck })); act(() => { eventBus.publish({ type: "run_input_ack", data: { id: "run-2" } } as WSMessage); eventBus.publish({ type: "run_input_ack", data: { id: "run-1" } } as WSMessage); }); expect(onInputAck).toHaveBeenCalledTimes(1); }); it("ignores every frame while the run id is null", () => { const onAnyStatus = vi.fn(); const { result } = renderHook(() => useRunStream(null, { ...noopOpts, onAnyStatus })); act(() => { eventBus.publish(streamMsg("run-1", { type: "result" })); eventBus.publish(statusMsg("run-1", "completed")); }); expect(result.current.envelopes).toEqual([]); expect(onAnyStatus).toHaveBeenCalledTimes(1); // id-agnostic by design }); it("disposes the event bus subscription on unmount", () => { const dispose = vi.fn(); const subscribe = vi.spyOn(eventBus, "subscribe").mockReturnValue(dispose); const { unmount } = renderHook(() => useRunStream("run-1", noopOpts)); expect(subscribe).toHaveBeenCalledTimes(1); expect(dispose).not.toHaveBeenCalled(); unmount(); expect(dispose).toHaveBeenCalledTimes(1); }); it("exposes setEnvelopes so the page can seed and clear the list", () => { const { result } = renderHook(() => useRunStream("run-1", noopOpts)); act(() => { result.current.setEnvelopes([{ type: "user", message: { content: "hello" } } as Envelope]); }); expect(result.current.envelopes).toHaveLength(1); act(() => { result.current.setEnvelopes([]); }); expect(result.current.envelopes).toEqual([]); }); });