diff --git a/apps/web/src/api/client.ts b/apps/web/src/api/client.ts index d7a80c4..807f188 100644 --- a/apps/web/src/api/client.ts +++ b/apps/web/src/api/client.ts @@ -42,7 +42,10 @@ export async function startScan(input: { }); } -export async function fetchCards(scanId: string): Promise<{ +export async function fetchCards( + scanId: string, + signal?: AbortSignal, +): Promise<{ scanId: string; status: string; cards: ApiCard[]; @@ -50,7 +53,7 @@ export async function fetchCards(scanId: string): Promise<{ driver?: string | null; recordingId?: string | null; }> { - return request(`/scans/${scanId}/cards`); + return request(`/scans/${scanId}/cards`, { signal }); } export async function postDecision( diff --git a/apps/web/src/hooks/useScanSession.test.ts b/apps/web/src/hooks/useScanSession.test.ts index 881c4dc..28670ad 100644 --- a/apps/web/src/hooks/useScanSession.test.ts +++ b/apps/web/src/hooks/useScanSession.test.ts @@ -1,6 +1,13 @@ import { describe, expect, it } from "vitest"; -import { applyEvent, emptyActivity } from "./useScanSession.js"; +import type { ApiCard } from "../api/types.js"; +import { + applyEvent, + createScanStartCoordinator, + emptyActivity, + reduce, + type ScanSessionState, +} from "./useScanSession.js"; const event = (type: string, systemId?: string) => ({ type, @@ -34,4 +41,113 @@ describe("scan session activity reducer", () => { expect(afterReconcile.subagents["failed-system"]?.status).toBe("failed"); expect(afterReconcile.subagents["healthy-system"]?.status).toBe("done"); }); + + it("discards a previous scan's card refresh after a new scan starts", () => { + const oldCard = { id: "old-card" } as ApiCard; + const newCard = { id: "new-card" } as ApiCard; + let state: ScanSessionState = { + activity: emptyActivity(), + cards: [] as ApiCard[], + loading: false, + error: null, + }; + + state = reduce(state, { type: "scan_starting", person: "Ada Lovelace" }); + state = reduce(state, { + type: "scan_started", + scanId: "scan-1", + person: "Ada Lovelace", + }); + const staleRefresh = { + type: "cards" as const, + scanId: "scan-1", + cards: [oldCard], + status: "completed", + }; + state = reduce(state, { type: "scan_starting", person: "Grace Hopper" }); + state = reduce(state, { + type: "scan_started", + scanId: "scan-2", + person: "Grace Hopper", + }); + + const staleResult = reduce(state, staleRefresh); + expect(staleResult.cards).toEqual([]); + expect(staleResult.activity.status).toBe("running"); + + const currentResult = reduce(staleResult, { + type: "cards", + scanId: "scan-2", + cards: [newCard], + status: "completed", + }); + expect(currentResult.cards).toEqual([newCard]); + expect(currentResult.activity.status).toBe("completed"); + + const lateStaleResult = reduce(currentResult, staleRefresh); + expect(lateStaleResult.cards).toEqual([newCard]); + expect(lateStaleResult.activity.status).toBe("completed"); + }); + + it("lets only the latest scan start own the SSE subscription and queue", () => { + const coordinator = createScanStartCoordinator(); + const existingUnsubscribe = { called: false }; + const existingToken = coordinator.begin(); + expect( + coordinator.commit(existingToken, "scan-0", () => { + existingUnsubscribe.called = true; + }), + ).toBe(true); + + const firstToken = coordinator.begin(); + const secondToken = coordinator.begin(); + const secondUnsubscribe = { called: false }; + expect( + coordinator.commit(secondToken, "scan-2", () => { + secondUnsubscribe.called = true; + }), + ).toBe(true); + expect(existingUnsubscribe.called).toBe(true); + expect(coordinator.activeScanId).toBe("scan-2"); + expect(coordinator.hasSubscription).toBe(true); + + const staleUnsubscribe = { called: false }; + expect( + coordinator.commit(firstToken, "scan-1", () => { + staleUnsubscribe.called = true; + }), + ).toBe(false); + expect(staleUnsubscribe.called).toBe(true); + expect(coordinator.activeScanId).toBe("scan-2"); + expect(secondUnsubscribe.called).toBe(false); + + let state: ScanSessionState = { + activity: emptyActivity(), + cards: [], + loading: false, + error: null, + }; + state = reduce(state, { type: "scan_starting", person: "Ada Lovelace" }); + state = reduce(state, { + type: "scan_started", + scanId: "scan-1", + person: "Ada Lovelace", + }); + state = reduce(state, { type: "scan_starting", person: "Grace Hopper" }); + state = reduce(state, { + type: "scan_started", + scanId: "scan-2", + person: "Grace Hopper", + }); + state = reduce(state, { + type: "cards", + scanId: "scan-1", + cards: [{ id: "stale" } as ApiCard], + status: "completed", + }); + + expect(state.activity.scanId).toBe("scan-2"); + expect(state.cards).toEqual([]); + expect(state.activity.status).toBe("running"); + }); }); diff --git a/apps/web/src/hooks/useScanSession.ts b/apps/web/src/hooks/useScanSession.ts index 7ca2e4b..94154d6 100644 --- a/apps/web/src/hooks/useScanSession.ts +++ b/apps/web/src/hooks/useScanSession.ts @@ -9,7 +9,7 @@ import type { } from "../api/types.js"; import { classifyClientError, recoveryFor } from "../lib/errors.js"; -type State = { +export type ScanSessionState = { activity: AgentActivityState; cards: ApiCard[]; loading: boolean; @@ -31,6 +31,7 @@ type Action = | { type: "event"; event: ScanProgressEvent } | { type: "cards"; + scanId: string; cards: ApiCard[]; status: string; costs?: AgentActivityState["costs"]; @@ -39,6 +40,55 @@ type Action = } | { type: "card_updated"; card: ApiCard }; +export interface ScanStartCoordinator { + begin(): number; + canCommit(token: number): boolean; + commit(token: number, scanId: string, unsubscribe: () => void): boolean; + cancel(): void; + readonly activeScanId: string | null; + readonly hasSubscription: boolean; +} + +export function createScanStartCoordinator(): ScanStartCoordinator { + let latestToken = 0; + let activeScanId: string | null = null; + let unsubscribe: (() => void) | null = null; + + return { + begin() { + latestToken += 1; + unsubscribe?.(); + unsubscribe = null; + activeScanId = null; + return latestToken; + }, + canCommit(token) { + return token === latestToken; + }, + commit(token, scanId, nextUnsubscribe) { + if (token !== latestToken) { + nextUnsubscribe(); + return false; + } + activeScanId = scanId; + unsubscribe = nextUnsubscribe; + return true; + }, + cancel() { + latestToken += 1; + unsubscribe?.(); + unsubscribe = null; + activeScanId = null; + }, + get activeScanId() { + return activeScanId; + }, + get hasSubscription() { + return unsubscribe !== null; + }, + }; +} + export const emptyActivity = (): AgentActivityState => ({ scanId: null, status: "idle", @@ -67,7 +117,7 @@ function pushLog( }; } -function reduce(state: State, action: Action): State { +export function reduce(state: ScanSessionState, action: Action): ScanSessionState { switch (action.type) { case "reset": return { activity: emptyActivity(), cards: [], loading: false, error: null }; @@ -130,6 +180,7 @@ function reduce(state: State, action: Action): State { }; } case "cards": + if (state.activity.scanId !== action.scanId) return state; return { ...state, cards: action.cards, @@ -460,29 +511,57 @@ export function useScanSession() { loading: false, error: null, }); - const unsubRef = useRef<(() => void) | null>(null); + const scanStartCoordinatorRef = useRef(null); + const refreshAbortRef = useRef(null); + if (scanStartCoordinatorRef.current === null) { + scanStartCoordinatorRef.current = createScanStartCoordinator(); + } + const scanStartCoordinator = scanStartCoordinatorRef.current; useEffect(() => { - return () => unsubRef.current?.(); - }, []); + return () => { + scanStartCoordinator.cancel(); + refreshAbortRef.current?.abort(); + }; + }, [scanStartCoordinator]); async function refreshCards(scanId: string) { - const res = await fetchCards(scanId); - dispatch({ - type: "cards", - cards: res.cards, - status: res.status, - costs: res.costs ?? null, - driver: (res.driver as string | null) ?? null, - recordingId: (res.recordingId as string | null) ?? null, - }); + refreshAbortRef.current?.abort(); + const controller = new AbortController(); + refreshAbortRef.current = controller; + try { + const res = await fetchCards(scanId, controller.signal); + if (scanStartCoordinator.activeScanId !== scanId || controller.signal.aborted) { + return; + } + dispatch({ + type: "cards", + scanId, + cards: res.cards, + status: res.status, + costs: res.costs ?? null, + driver: (res.driver as string | null) ?? null, + recordingId: (res.recordingId as string | null) ?? null, + }); + } catch (err) { + if (!controller.signal.aborted && scanStartCoordinator.activeScanId === scanId) { + throw err; + } + } finally { + if (refreshAbortRef.current === controller) { + refreshAbortRef.current = null; + } + } } async function beginScan(person: string) { - unsubRef.current?.(); + const startToken = scanStartCoordinator.begin(); + refreshAbortRef.current?.abort(); + refreshAbortRef.current = null; dispatch({ type: "scan_starting", person }); try { const started = await startScan({ person }); + if (!scanStartCoordinator.canCommit(startToken)) return; dispatch({ type: "scan_started", scanId: started.scanId, @@ -490,7 +569,7 @@ export function useScanSession() { driver: started.driver, recordingId: started.recordingId ?? null, }); - unsubRef.current = subscribeScanStream(started.scanId, { + const unsubscribe = subscribeScanStream(started.scanId, { onEvent: (event) => { dispatch({ type: "event", event }); if ( @@ -505,9 +584,13 @@ export function useScanSession() { } }, }); + if (!scanStartCoordinator.commit(startToken, started.scanId, unsubscribe)) { + return; + } // Initial poll in case events already finished void refreshCards(started.scanId); } catch (err) { + if (!scanStartCoordinator.canCommit(startToken)) return; dispatch({ type: "scan_error", error: err instanceof Error ? err.message : String(err),