From fca5808f5b973213aa00200ce83a2e77c7721d1d Mon Sep 17 00:00:00 2001 From: Brad Harris Date: Fri, 14 Aug 2026 20:33:49 -0600 Subject: [PATCH] Cover the SSE event dispatch table in use-sse MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reconnect state machine added in #921 was already well covered, but `handleSSEMessage` — the table that routes ~20 server push types into react-query cache writes — was not. Only `snapshot` was exercised, and only as a delivery vehicle for the backoff tests, with nothing asserted about what it does to the cache. Every realtime update in the app flows through this function. Adds 23 tests driving the real hook through the fake EventSource, so the routing itself is what gets pinned: which payload reaches which key, the `exact` scoping that keeps one agent's event from refetching every other agent's list, the review-detail invalidation predicate, the ack that must only fire when a notification was actually shown, and the tolerance of an unparseable or unrecognized frame. Also covers the connection gate — the stream must not open before authentication, including on the foreground path, which reopens without re-running the effect. Mutation battery: 50 mutants, 47 killed. The three survivors are provably unobservable rather than uncovered — `patchAgentHasStream`'s no-op guard and `media.seen`'s `!file.seen` guard are each redundant with react-query structural sharing, and `applyReviewCreated`'s null guard only ever skips an element-identical rewrite. An earlier draft asserted referential stability to chase the first two; those assertions were dropped because no change to use-sse.ts can fail them, which made them a test of react-query rather than of this hook. Co-Authored-By: Claude Opus 5 (1M context) --- apps/web/src/hooks/use-sse.test.ts | 578 ++++++++++++++++++++++++++++- 1 file changed, 575 insertions(+), 3 deletions(-) diff --git a/apps/web/src/hooks/use-sse.test.ts b/apps/web/src/hooks/use-sse.test.ts index f6d0c4e1..07ba0ecc 100644 --- a/apps/web/src/hooks/use-sse.test.ts +++ b/apps/web/src/hooks/use-sse.test.ts @@ -2,12 +2,20 @@ import { createElement, type ReactNode } from "react"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { act, cleanup, renderHook } from "@testing-library/react"; +import { createStore, Provider } from "jotai"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { agentDiffQueryKey } from "@/hooks/use-agent-diff"; import { diffStatsQueryKey } from "@/hooks/use-agent-diff-stats"; +import { CACHED_RELEASE_INFO_QUERY_KEY } from "@/hooks/use-cached-release-info"; +import { whiteboardAgentDrewAtomFamily } from "@/lib/store"; +import { showWebNotification } from "@/lib/web-notifications"; -import { type Agent } from "@/components/app/types"; +import { + type Agent, + type AuthState, + type MediaFile, +} from "@/components/app/types"; import { applyAgentUpsert, @@ -16,6 +24,10 @@ import { useSSE, } from "./use-sse"; +vi.mock("@/lib/web-notifications", () => ({ + showWebNotification: vi.fn(() => false), +})); + function agent( id: string, submittedReviewId: number | null, @@ -139,9 +151,9 @@ class FakeEventSource { describe("useSSE reconnect", () => { let hiddenValue = false; - function renderSSE() { + function renderSSE(authState: AuthState = "authenticated") { const queryClient = new QueryClient(); - return renderHook(() => useSSE("authenticated"), { + return renderHook(() => useSSE(authState), { wrapper: ({ children }: { children: ReactNode }) => createElement(QueryClientProvider, { client: queryClient }, children), }); @@ -308,4 +320,564 @@ describe("useSSE reconnect", () => { expect(FakeEventSource.instances).toHaveLength(1); }); + + it("does not open a stream until the session is authenticated", () => { + // The stream is credentialed; opening it before login just earns a 401 + // that the fatal-error path would then retry against forever. + const { rerender } = renderHook( + ({ authState }: { authState: AuthState }) => useSSE(authState), + { + initialProps: { authState: "loading" as AuthState }, + wrapper: ({ children }: { children: ReactNode }) => + createElement( + QueryClientProvider, + { client: new QueryClient() }, + children + ), + } + ); + expect(FakeEventSource.instances).toHaveLength(0); + + act(() => void vi.advanceTimersByTime(60_000)); + expect(FakeEventSource.instances).toHaveLength(0); + + // Foregrounding is the one path that opens a stream without re-running the + // effect, so it has to re-check auth rather than assume it. + act(() => setHidden(true)); + act(() => setHidden(false)); + expect(FakeEventSource.instances).toHaveLength(0); + + rerender({ authState: "authenticated" as AuthState }); + expect(FakeEventSource.instances).toHaveLength(1); + }); + + it("does not open a stream while the tab starts hidden", () => { + hiddenValue = true; + renderSSE(); + + expect(FakeEventSource.instances).toHaveLength(0); + + act(() => setHidden(false)); + expect(FakeEventSource.instances).toHaveLength(1); + }); +}); + +/** + * The message handler is the wiring hub for the whole realtime UI: every + * server push lands here and is translated into a specific cache mutation or + * invalidation. Drive the real hook through the fake stream rather than + * calling helpers directly, so the routing itself — which payload reaches + * which key — is what gets pinned. + */ +describe("useSSE message handling", () => { + function renderMessages() { + const queryClient = new QueryClient(); + const jotaiStore = createStore(); + const invalidateQueries = vi.spyOn(queryClient, "invalidateQueries"); + const removeQueries = vi.spyOn(queryClient, "removeQueries"); + + const utils = renderHook(() => useSSE("authenticated"), { + wrapper: ({ children }: { children: ReactNode }) => + createElement( + Provider, + { store: jotaiStore }, + createElement(QueryClientProvider, { client: queryClient }, children) + ), + }); + + const source = FakeEventSource.instances[0]!; + const emit = (payload: unknown) => act(() => source.emit(payload)); + /** Push a raw (possibly unparseable) frame the way the server would. */ + const emitRaw = (data: string) => + act(() => { + source.onmessage?.(new MessageEvent("message", { data })); + }); + + return { + queryClient, + jotaiStore, + invalidateQueries, + removeQueries, + emit, + emitRaw, + ...utils, + }; + } + + /** Every queryKey passed to invalidateQueries, in call order. */ + function invalidatedKeys(spy: { mock: { calls: unknown[][] } }): unknown[] { + return spy.mock.calls + .map( + ([filters]) => (filters as { queryKey?: unknown } | undefined)?.queryKey + ) + .filter((key) => key !== undefined); + } + + /** + * Assert exactly this set of keys was invalidated — nothing missing, nothing + * extra — without pinning the order. Within a single event the order the + * branch happens to issue its invalidations in is not a contract, so + * reordering them should stay a behavior-neutral refactor. + */ + function expectInvalidatedSet( + spy: { mock: { calls: unknown[][] } }, + expected: unknown[] + ): void { + const keys = invalidatedKeys(spy); + expect(keys).toHaveLength(expected.length); + expect(keys).toEqual(expect.arrayContaining(expected)); + } + + let fetchMock: ReturnType; + + beforeEach(() => { + FakeEventSource.instances = []; + vi.stubGlobal("EventSource", FakeEventSource); + fetchMock = vi.fn(() => Promise.resolve({} as Response)); + vi.stubGlobal("fetch", fetchMock); + // vitest's restoreAllMocks only resets vi.spyOn spies, so a module-factory + // vi.fn() keeps its call history and implementation across tests unless it + // is reset explicitly here. + vi.mocked(showWebNotification).mockReset(); + vi.mocked(showWebNotification).mockReturnValue(false); + }); + + afterEach(() => { + // This config sets neither `globals: true` nor `setupFiles`, so RTL's + // auto-cleanup never registers — see the note in the reconnect suite. + cleanup(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it("replaces the agent list from a snapshot in created-at order", () => { + const { queryClient, emit } = renderMessages(); + queryClient.setQueryData(["agents"], [agent("stale", null)]); + + emit({ + type: "snapshot", + agents: [ + agent("older", null, "2026-07-16T10:00:00.000Z"), + agent("newer", null, "2026-07-16T14:00:00.000Z"), + ], + }); + + expect( + queryClient.getQueryData(["agents"])?.map((a) => a.id) + ).toEqual(["newer", "older"]); + }); + + it("refetches the state a snapshot does not carry and drops injection holds", () => { + // The snapshot payload only carries agents. Everything else the UI holds + // could have changed during the gap, and injection-hold state is + // event-sourced with no endpoint to refetch — so it has to fail safe. + const { emit, invalidateQueries, removeQueries } = renderMessages(); + + emit({ type: "snapshot", agents: [] }); + + expectInvalidatedSet(invalidateQueries, [ + ["jobs"], + ["templates"], + ["brain"], + ["whiteboard"], + CACHED_RELEASE_INFO_QUERY_KEY, + ]); + expect(removeQueries).toHaveBeenCalledWith({ + queryKey: ["injection-hold"], + }); + }); + + it("inserts an upserted agent in sorted position", () => { + const { queryClient, emit } = renderMessages(); + queryClient.setQueryData( + ["agents"], + [agent("old", null, "2026-07-16T10:00:00.000Z")] + ); + + emit({ + type: "agent.upsert", + agent: agent("fresh", null, "2026-07-16T18:00:00.000Z"), + }); + + expect( + queryClient.getQueryData(["agents"])?.map((a) => a.id) + ).toEqual(["fresh", "old"]); + }); + + it("removes only the deleted agent, and empties an unseeded list", () => { + const { queryClient, emit } = renderMessages(); + queryClient.setQueryData( + ["agents"], + [agent("keep", null), agent("drop", null)] + ); + + emit({ type: "agent.deleted", agentId: "drop" }); + + expect( + queryClient.getQueryData(["agents"])?.map((a) => a.id) + ).toEqual(["keep"]); + + // A delete for an agent list this tab never fetched must still leave a + // defined (empty) list rather than writing undefined back over the key. + queryClient.removeQueries({ queryKey: ["agents"] }); + emit({ type: "agent.deleted", agentId: "drop" }); + expect(queryClient.getQueryData(["agents"])).toEqual([]); + }); + + it("writes terminal and injection-hold state under their agent keys", () => { + const { queryClient, emit } = renderMessages(); + + emit({ + type: "agent.terminal_state_changed", + agentId: "a1", + terminalState: { copyMode: "copy", lastObservedAt: 7 }, + }); + emit({ + type: "agent.injection_hold_changed", + agentId: "a2", + holdState: { held: true, pendingCount: 3, quietMs: 900 }, + }); + + expect(queryClient.getQueryData(["terminal-state", "a1"])).toEqual({ + copyMode: "copy", + lastObservedAt: 7, + }); + expect(queryClient.getQueryData(["injection-hold", "a2"])).toEqual({ + held: true, + pendingCount: 3, + quietMs: 900, + }); + }); + + it("routes diff state to the include-uncommitted stats key", () => { + const { queryClient, emit } = renderMessages(); + const pushed = { added: 3, deleted: 4, files: 2, computedAt: 123 }; + + emit({ + type: "agent.diff_state_changed", + agentId: "a1", + diffStats: pushed, + }); + + expect(queryClient.getQueryData(diffStatsQueryKey("a1", true))).toEqual( + pushed + ); + }); + + it("flips hasStream on the streaming agent only", () => { + const { queryClient, emit } = renderMessages(); + queryClient.setQueryData( + ["agents"], + [ + { ...agent("a1", null), hasStream: false } as Agent, + { ...agent("a2", null), hasStream: false } as Agent, + ] + ); + + emit({ type: "stream.started", agentId: "a1" }); + expect( + queryClient + .getQueryData(["agents"]) + ?.map((a) => [a.id, a.hasStream]) + ).toEqual([ + ["a1", true], + ["a2", false], + ]); + + emit({ type: "stream.stopped", agentId: "a1" }); + expect( + queryClient + .getQueryData(["agents"]) + ?.map((a) => [a.id, a.hasStream]) + ).toEqual([ + ["a1", false], + ["a2", false], + ]); + }); + + it("marks the agent-drew flag only when the agent did the drawing", () => { + // The flag drives an attention affordance, so echoing the user's own + // strokes back at them would make it permanently lit. + const { jotaiStore, emit, invalidateQueries } = renderMessages(); + + emit({ + type: "whiteboard.changed", + agentId: "a1", + version: 2, + source: "user", + }); + expect(jotaiStore.get(whiteboardAgentDrewAtomFamily("a1"))).toBe(false); + expect(invalidateQueries).toHaveBeenCalledWith({ + queryKey: ["whiteboard", "a1"], + exact: true, + }); + + emit({ + type: "whiteboard.changed", + agentId: "a1", + version: 3, + source: "agent", + }); + expect(jotaiStore.get(whiteboardAgentDrewAtomFamily("a1"))).toBe(true); + // Another agent's board is untouched by a1's stroke. + expect(jotaiStore.get(whiteboardAgentDrewAtomFamily("a2"))).toBe(false); + }); + + it("marks seen only the media files named in the event", () => { + const { queryClient, emit } = renderMessages(); + const file = (name: string, updatedAt: string): MediaFile => ({ + name, + updatedAt, + size: 1, + url: `/media/${name}`, + }); + queryClient.setQueryData( + ["media", "a1"], + [ + file("shot.png", "2026-07-16T10:00:00.000Z"), + file("other.png", "2026-07-16T10:00:00.000Z"), + // Same name, newer revision — the key is name+updatedAt, so a stale + // seen-key must not mark the replacement read. + file("shot.png", "2026-07-16T11:00:00.000Z"), + ] + ); + + emit({ + type: "media.seen", + agentId: "a1", + keys: ["shot.png:2026-07-16T10:00:00.000Z"], + }); + + expect( + queryClient + .getQueryData(["media", "a1"]) + ?.map((f) => f.seen ?? false) + ).toEqual([true, false, false]); + }); + + it("invalidates one agent's media list on media.changed", () => { + const { emit, invalidateQueries } = renderMessages(); + + emit({ type: "media.changed", agentId: "a1" }); + + // Exact, or every other agent's media list refetches on one screenshot. + expect(invalidateQueries).toHaveBeenCalledWith({ + queryKey: ["media", "a1"], + exact: true, + }); + }); + + it("marks the reviewer submitted and refreshes review state on review.created", () => { + const { queryClient, emit, invalidateQueries } = renderMessages(); + queryClient.setQueryData( + ["agents"], + [agent("reviewer", null), agent("author", null)] + ); + + emit({ + type: "review.created", + agentId: "author", + reviewId: 42, + reviewerAgentId: "reviewer", + }); + + expect( + queryClient + .getQueryData(["agents"]) + ?.map((a) => a.submittedReviewId) + ).toEqual([42, null]); + expectInvalidatedSet(invalidateQueries, [ + ["agent-reviews", "author"], + ["agent-feedback-items", "author"], + ]); + }); + + it("still refreshes review state when no reviewer is attributed", () => { + // Human-authored reviews carry no reviewer agent. The missing attribution + // must skip only the badge, not the refresh of the review lists. + const { queryClient, emit, invalidateQueries } = renderMessages(); + queryClient.setQueryData(["agents"], [agent("author", null)]); + + emit({ + type: "review.created", + agentId: "author", + reviewId: 42, + reviewerAgentId: null, + }); + + expect( + queryClient.getQueryData(["agents"])?.[0]?.submittedReviewId + ).toBeNull(); + expectInvalidatedSet(invalidateQueries, [ + ["agent-reviews", "author"], + ["agent-feedback-items", "author"], + ]); + }); + + it("scopes the review-detail predicate to the event's agent", () => { + const { emit, invalidateQueries } = renderMessages(); + + emit({ type: "review.updated", agentId: "author" }); + + const predicate = invalidateQueries.mock.calls + .map( + ([filters]) => + (filters as { predicate?: (q: unknown) => boolean }).predicate + ) + .find(Boolean)!; + const matches = (queryKey: unknown[]) => predicate({ queryKey } as never); + + expect(matches(["agent-review-detail", "author", 1])).toBe(true); + expect(matches(["agent-review-detail", "someone-else", 1])).toBe(false); + expect(matches(["agent-reviews", "author"])).toBe(false); + }); + + it("leaves submittedReviewId alone for review updates", () => { + // Only creation carries a reviewId; an update must not clear or reassign + // the badge the reviewer already earned. + const { queryClient, emit } = renderMessages(); + queryClient.setQueryData(["agents"], [agent("reviewer", 42)]); + + emit({ type: "review_feedback.updated", agentId: "reviewer" }); + + expect( + queryClient.getQueryData(["agents"])?.[0]?.submittedReviewId + ).toBe(42); + }); + + it("invalidates both ends of a created message and only one on read", () => { + const { emit, invalidateQueries } = renderMessages(); + + emit({ + type: "message.created", + senderAgentId: "sender", + recipientAgentId: "recipient", + }); + expectInvalidatedSet(invalidateQueries, [ + ["messages", "sender"], + ["messages", "recipient"], + ]); + + invalidateQueries.mockClear(); + emit({ type: "message.read", agentId: "recipient" }); + expect(invalidateQueries).toHaveBeenCalledWith({ + queryKey: ["messages", "recipient"], + exact: true, + }); + }); + + it("invalidates the collection each bare change event names", () => { + const { emit, invalidateQueries } = renderMessages(); + + emit({ type: "job.changed" }); + emit({ type: "template.changed" }); + emit({ type: "brain.changed", repoRoot: "/repo" }); + + // Ordered on purpose here, unlike the within-one-event assertions above: + // the sequence is the test's own emit order, so it is what proves each + // event invalidates its own collection instead of all three sharing one. + expect(invalidatedKeys(invalidateQueries)).toEqual([ + ["jobs"], + ["templates"], + ["brain"], + ]); + }); + + it("acks a notification only when one was actually shown", () => { + const { emit } = renderMessages(); + const payload = { + type: "notification", + notificationId: "n1", + agentId: "a1", + agentName: "Agent One", + eventType: "done", + message: "finished", + }; + + // Permission denied — nothing was displayed, so the server must keep the + // notification queued for whichever tab can show it. + emit(payload); + expect(showWebNotification).toHaveBeenCalledWith( + expect.objectContaining({ agentName: "Agent One", eventType: "done" }) + ); + expect(fetchMock).not.toHaveBeenCalled(); + + vi.mocked(showWebNotification).mockReturnValue(true); + emit(payload); + expect(fetchMock).toHaveBeenCalledWith( + "/api/v1/notifications/ack", + expect.objectContaining({ + method: "POST", + credentials: "include", + // keepalive, or the ack is cancelled when the click navigates away. + keepalive: true, + body: JSON.stringify({ notificationId: "n1" }), + }) + ); + }); + + it("handles the rejection of a fire-and-forget ack", async () => { + // The ack is never awaited, so a failed one — offline, server restarting — + // has to be swallowed at the call site or it surfaces as an unhandled + // rejection in the user's console. Assert the handler is attached rather + // than that nothing threw: the whole dispatch runs inside a try/catch, so + // a not-toThrow assertion here would pass with the .catch() deleted. + const { emit } = renderMessages(); + vi.mocked(showWebNotification).mockReturnValue(true); + const rejection = Promise.reject(new Error("offline")); + const attachCatch = vi.spyOn(rejection, "catch"); + fetchMock.mockReturnValueOnce(rejection); + + emit({ + type: "notification", + notificationId: "n1", + agentId: "a1", + agentName: "Agent One", + eventType: "done", + message: "finished", + }); + + expect(attachCatch).toHaveBeenCalled(); + await act(async () => {}); + rejection.catch(() => {}); + }); + + it("caches a pushed release snapshot under the response shape", () => { + const { queryClient, emit } = renderMessages(); + const snapshot = { currentTag: "v1.0.0", updateAvailable: true }; + + emit({ type: "release.cached_info_changed", snapshot }); + + // The query's own fetch returns { snapshot }, so the push has to match or + // consumers read undefined off a populated cache. + expect(queryClient.getQueryData(CACHED_RELEASE_INFO_QUERY_KEY)).toEqual({ + snapshot, + }); + }); + + it("ignores an unparseable frame and keeps handling the next one", () => { + const { queryClient, emit, emitRaw } = renderMessages(); + queryClient.setQueryData(["agents"], [agent("keep", null)]); + + expect(() => emitRaw("not json")).not.toThrow(); + expect( + queryClient.getQueryData(["agents"])?.map((a) => a.id) + ).toEqual(["keep"]); + + // A poisoned frame must not wedge the stream. + emit({ type: "agent.deleted", agentId: "keep" }); + expect(queryClient.getQueryData(["agents"])).toEqual([]); + }); + + it("ignores an event type it does not know", () => { + const { queryClient, emit, invalidateQueries } = renderMessages(); + queryClient.setQueryData(["agents"], [agent("keep", null)]); + + emit({ type: "agent.invented_by_a_newer_server", agentId: "keep" }); + + expect( + queryClient.getQueryData(["agents"])?.map((a) => a.id) + ).toEqual(["keep"]); + expect(invalidateQueries).not.toHaveBeenCalled(); + }); });