diff --git a/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.test.tsx b/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.test.tsx new file mode 100644 index 0000000000..8a90be8b2e --- /dev/null +++ b/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.test.tsx @@ -0,0 +1,122 @@ +import { act, renderHook } from "@testing-library/react"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { RunEvent } from "@/lib/api/domains/office-extended-api"; +import type { RunEventAppendedPayload } from "@/lib/types/backend"; +import { useRunLiveSync } from "./use-run-live-sync"; + +const clients = vi.hoisted(() => ({ + active: { subscribeRun: vi.fn() }, +})); + +vi.mock("@/lib/ws/connection", () => ({ + getWebSocketClient: () => clients.active, + useWebSocketClient: () => clients.active, +})); + +const handlers = vi.hoisted(() => ({ + listener: undefined as ((payload: RunEventAppendedPayload) => void) | undefined, +})); + +vi.mock("@/lib/ws/handlers/run", () => ({ + subscribeRunEvents: (_runId: string, listener: (payload: RunEventAppendedPayload) => void) => { + handlers.listener = listener; + return () => { + handlers.listener = undefined; + }; + }, +})); + +function runEvent(seq: number, eventType: string): RunEvent { + return { + seq, + event_type: eventType, + level: "info", + payload: "{}", + created_at: "2026-01-01T00:00:00Z", + }; +} + +type Props = { + runId: string; + initialEvents: RunEvent[]; + initialStatus: "claimed"; +}; + +function renderLiveSync(initialProps: Props) { + return renderHook( + (props: Props) => useRunLiveSync(props.runId, props.initialEvents, props.initialStatus), + { initialProps }, + ); +} + +const initialProps: Props = { runId: "run-1", initialEvents: [], initialStatus: "claimed" }; + +describe("useRunLiveSync", () => { + beforeEach(() => { + clients.active = { subscribeRun: vi.fn(() => vi.fn()) }; + handlers.listener = undefined; + }); + + it("does not loop when the snapshot reference changes but its content does not", () => { + // A fresh array literal on every render changes `initialEvents` identity + // without changing its content. Without the content guard the snapshot + // sync effect re-writes state on every render, feeding an endless + // render->effect cycle that exhausts the worker (this test times out + // rather than completing). + const { result, rerender } = renderHook(() => useRunLiveSync("run-1", [], "claimed")); + + rerender(); + rerender(); + + expect(clients.active.subscribeRun).toHaveBeenCalledOnce(); + expect(result.current.events).toEqual([]); + expect(result.current.status).toBe("claimed"); + }); + + it("does not clobber a live event when a fresh but unchanged snapshot rerenders", () => { + // A caller that supplies a fresh-but-unchanged `initialEvents` array (the + // exact scenario the snapshot guard exists for) must not erase events that + // arrived over the WS. The snapshot sync decision has to compare against + // the last incoming snapshot, not against the live-event dedup set: once + // the first WS event adds its seq to the dedup set, comparing the stale + // snapshot against that set looks like a content change and resets events + // (and the dedup set) to the snapshot, making the live event vanish. + const { result, rerender } = renderLiveSync(initialProps); + + const liveEvent = runEvent(1, "started"); + act(() => { + handlers.listener!({ run_id: "run-1", event: liveEvent }); + }); + expect(result.current.events).toEqual([liveEvent]); + + rerender({ ...initialProps, initialEvents: [] }); + + expect(result.current.events).toEqual([liveEvent]); + }); + + it("merges a changed snapshot without dropping a live event", () => { + const initialEvent = runEvent(1, "init"); + const liveEvent = runEvent(2, "started"); + const snapshotEvent = runEvent(3, "step"); + const { result, rerender } = renderLiveSync({ + ...initialProps, + initialEvents: [initialEvent], + }); + + act(() => { + handlers.listener!({ run_id: "run-1", event: liveEvent }); + }); + rerender({ ...initialProps, initialEvents: [initialEvent, snapshotEvent] }); + + expect(result.current.events).toEqual([initialEvent, liveEvent, snapshotEvent]); + }); + + it("re-syncs events when the snapshot content actually changes", () => { + const started = runEvent(1, "started"); + const { result, rerender } = renderLiveSync(initialProps); + + rerender({ ...initialProps, initialEvents: [started] }); + + expect(result.current.events).toEqual([started]); + }); +}); diff --git a/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.ts b/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.ts index 2778869535..773672b729 100644 --- a/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.ts +++ b/apps/web/app/office/agents/[id]/runs/[runId]/use-run-live-sync.ts @@ -40,12 +40,46 @@ export function useRunLiveSync( ): { events: RunEvent[]; status: Status } { const [events, setEvents] = useState(initialEvents); const [status, setStatus] = useState(initialStatus); + // Live-event dedup only: seqs of events already appended over the WS (plus + // the seqs of the mount-time snapshot, so a reconnect replay of a + // snapshot-covered event cannot double-insert). Never reset to a snapshot: + // doing so erases live events on a fresh-but-unchanged rerender. const seenSeqsRef = useRef>(new Set(initialEvents.map((e) => e.seq))); + // Last incoming snapshot, tracked separately from the live dedup set. The + // sync decision compares against this, so live-added seqs can never make + // an unchanged snapshot look like a content change. + const lastSnapshotRef = useRef<{ runId: string; seqs: Set }>({ + runId, + seqs: new Set(initialEvents.map((e) => e.seq)), + }); useEffect(() => { - seenSeqsRef.current = new Set(initialEvents.map((e) => e.seq)); + const nextSeqs = new Set(initialEvents.map((e) => e.seq)); + const prev = lastSnapshotRef.current; + if ( + prev.runId === runId && + prev.seqs.size === nextSeqs.size && + [...nextSeqs].every((seq) => prev.seqs.has(seq)) + ) { + // Same run, same event set as the last incoming snapshot. Skip the state + // write so an unstable `initialEvents` reference (a fresh array literal + // on every render) cannot feed an endless render->effect cycle, and so a + // stale snapshot cannot clobber events appended live over the WS. + return; + } + lastSnapshotRef.current = { runId, seqs: nextSeqs }; + if (prev.runId === runId) { + // Fresh snapshot for the same run: merge it with live-added events and + // fold its seqs into the dedup set so reconnect replays cannot duplicate. + for (const seq of nextSeqs) seenSeqsRef.current.add(seq); + setEvents((current) => mergeRunEvents(current, initialEvents)); + return; + } else { + // Different run: restart dedup from the new snapshot; seqs are per-run. + seenSeqsRef.current = new Set(nextSeqs); + } setEvents(initialEvents); - }, [initialEvents]); + }, [initialEvents, runId]); useEffect(() => { setStatus(initialStatus); @@ -97,3 +131,13 @@ function mergeRunEvent(prev: RunEvent[], evt: RunEvent): RunEvent[] { next.splice(lo, 0, evt); return next; } + +function mergeRunEvents(current: RunEvent[], snapshot: RunEvent[]): RunEvent[] { + if (current.length === 0) return snapshot; + if (snapshot.length === 0) return current; + + const bySeq = new Map(); + for (const event of current) bySeq.set(event.seq, event); + for (const event of snapshot) bySeq.set(event.seq, event); + return [...bySeq.values()].sort((a, b) => a.seq - b.seq); +}