import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import type { HeartbeatRunEvent } from "@paperclipai/shared"; import type { TranscriptEntry } from "@/adapters"; import { heartbeatsApi } from "@/api/heartbeats"; import { nativeRunEventsToTranscript } from "./native-run-events"; import { readTranscriptRequest } from "./read-transcript-request"; const EVENT_PAGE_SIZE = 1_000; const EVENT_POLL_INTERVAL_MS = 2_000; export interface NativeRunTranscriptSource { id: string; status: string; runtimeMode?: "legacy" | "native"; } export interface NativeRunTranscriptError { message: string; failedAt: string; } function isLive(status: string): boolean { return status === "queued" || status === "running"; } export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource[]) { const nativeRunsKey = runs .filter((run) => run.runtimeMode === "native") .map((run) => `${run.id}:${run.status}`) .sort() .join(","); const nativeRuns = useMemo( () => runs.filter((run) => run.runtimeMode === "native").map((run) => ({ ...run })), // The key carries every field this hook consumes. // eslint-disable-next-line react-hooks/exhaustive-deps [nativeRunsKey], ); const [eventsByRun, setEventsByRun] = useState>(new Map()); const [errorsByRun, setErrorsByRun] = useState>(new Map()); const [hydratedRunIds, setHydratedRunIds] = useState>(new Set()); const [hydratedStatusByRun, setHydratedStatusByRun] = useState(new Map()); const [retryGeneration, setRetryGeneration] = useState(0); const retry = useCallback(() => setRetryGeneration((value) => value + 1), []); const projectionCacheRef = useRef(new Map()); const cursorByRunRef = useRef(new Map()); useEffect(() => { let cancelled = false; const controller = new AbortController(); const timers = new Set(); const retainedIds = new Set(nativeRuns.map((run) => run.id)); const retainMap = (previous: Map) => { const next = new Map([...previous].filter(([id]) => retainedIds.has(id))); return next.size === previous.size ? previous : next; }; setEventsByRun(retainMap); setErrorsByRun(retainMap); setHydratedStatusByRun(retainMap); setHydratedRunIds((previous) => { const next = new Set([...previous].filter((id) => retainedIds.has(id))); return next.size === previous.size ? previous : next; }); for (const id of cursorByRunRef.current.keys()) { if (!retainedIds.has(id)) cursorByRunRef.current.delete(id); } const refreshRun = async (run: NativeRunTranscriptSource) => { let failed = false; try { let cursor = cursorByRunRef.current.get(run.id) ?? 0; const incoming: HeartbeatRunEvent[] = []; for (;;) { const page = await readTranscriptRequest( (signal) => heartbeatsApi.events(run.id, cursor, EVENT_PAGE_SIZE, { signal }), controller.signal, ); if (cancelled) return; const last = page.at(-1); const nextCursor = last ? Math.max(cursor, last.seq) : cursor; incoming.push(...page.filter((event) => event.seq > cursor)); if (page.length < EVENT_PAGE_SIZE || nextCursor === cursor) { cursor = nextCursor; break; } cursor = nextCursor; } // Commit this run's cursor with its rows. A slow sibling must neither // hold its readiness hostage nor stall live polling for this run. cursorByRunRef.current.set(run.id, cursor); if (incoming.length > 0) setEventsByRun((previous) => { const next = new Map(previous); next.set(run.id, [...(previous.get(run.id) ?? []), ...incoming]); return next; }); setErrorsByRun((previous) => { if (!previous.has(run.id)) return previous; const next = new Map(previous); next.delete(run.id); return next; }); setHydratedStatusByRun((previous) => { if (previous.get(run.id) === run.status) return previous; return new Map(previous).set(run.id, run.status); }); } catch (error) { if (cancelled) return; failed = true; setErrorsByRun((previous) => { if (previous.has(run.id)) return previous; const next = new Map(previous); next.set(run.id, { message: error instanceof Error ? error.message : "Native run activity could not be loaded", failedAt: new Date().toISOString(), }); return next; }); } if (cancelled) return; setHydratedRunIds((previous) => previous.has(run.id) ? previous : new Set([...previous, run.id])); if (isLive(run.status) || failed) { const timer = window.setTimeout(() => { timers.delete(timer); void refreshRun(run); }, EVENT_POLL_INTERVAL_MS); timers.add(timer); } }; for (const run of nativeRuns) void refreshRun(run); return () => { cancelled = true; controller.abort(); for (const timer of timers) window.clearTimeout(timer); }; }, [nativeRuns, retryGeneration]); const transcriptByRun = useMemo(() => { const transcripts = new Map(); for (const run of nativeRuns) { const events = eventsByRun.get(run.id); if (!events) continue; let cached = projectionCacheRef.current.get(run.id); if (!cached || cached.events !== events) { cached = { events, transcript: nativeRunEventsToTranscript(events) }; projectionCacheRef.current.set(run.id, cached); } transcripts.set(run.id, cached.transcript); } for (const id of projectionCacheRef.current.keys()) { if (!transcripts.has(id)) projectionCacheRef.current.delete(id); } return transcripts; }, [eventsByRun, nativeRuns]); const terminalHydratedRunIds = useMemo(() => new Set(nativeRuns .filter((run) => !isLive(run.status) && hydratedStatusByRun.get(run.id) === run.status) .map((run) => run.id)), [nativeRuns, hydratedStatusByRun]); return { transcriptByRun, errorsByRun, hydratedRunIds, terminalHydratedRunIds, retry, isInitialHydrating: nativeRuns.some((run) => !hydratedRunIds.has(run.id)), }; }