fix(ui): stabilize task loading and live feeds (#13095)

Coordinate initial conversation reveal, preserve message identity and reading anchors during live updates, and bound transcript reads with recoverable retries. Cover desktop/mobile navigation and rich task loading with actual-route browser tests.

Verified all Linux CI gates, 5,575 local UI tests, eight layout browser scenarios, and recorded native Codex walkthroughs. Greptile: 5/5; all review findings resolved.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

ui/src/components/transcript/read-transcript-request.ts

@@ -0,0 +1,35 @@
+export const TRANSCRIPT_REQUEST_TIMEOUT_MS = 15_000;
+
+/** Bound history reads, including body consumption, and release coalesced GETs
+ * on timeout so Retry starts a fresh request. Late responses cannot commit. */
+export function readTranscriptRequest<T>(request: (signal: AbortSignal) => Promise<T>, signal: AbortSignal): Promise<T> {
+  if (signal.aborted) return Promise.reject(new DOMException("History read cancelled", "AbortError"));
+  return new Promise<T>((resolve, reject) => {
+    const controller = new AbortController();
+    let settled = false;
+    const finish = (complete: () => void) => {
+      if (settled) return;
+      settled = true;
+      window.clearTimeout(timer);
+      signal.removeEventListener("abort", onAbort);
+      complete();
+    };
+    const onAbort = () => {
+      finish(() => reject(new DOMException("History read cancelled", "AbortError")));
+      controller.abort();
+    };
+    const timer = window.setTimeout(() => {
+      finish(() => reject(new Error("Run history took too long to load. Retry to load it.")));
+      controller.abort();
+    }, TRANSCRIPT_REQUEST_TIMEOUT_MS);
+    signal.addEventListener("abort", onAbort, { once: true });
+    try {
+      request(controller.signal).then(
+        (value) => finish(() => resolve(value)),
+        (error) => finish(() => reject(error)),
+      );
+    } catch (error) {
+      finish(() => reject(error));
+    }
+  });
+}

ui/src/components/transcript/useLiveRunTranscripts.test.tsx

@@ -5,6 +5,7 @@ import { createRoot } from "react-dom/client";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { ApiError } from "../../api/client";
 import { useLiveRunTranscripts } from "./useLiveRunTranscripts";
+import { TRANSCRIPT_REQUEST_TIMEOUT_MS } from "./read-transcript-request";
 
 const { useQueryMock, logMock, buildTranscriptMock } = vi.hoisted(() => ({
   useQueryMock: vi.fn(() => ({ data: { censorUsernameInLogs: false } })),
@@ -198,6 +199,36 @@ describe("useLiveRunTranscripts", () => {
     container.remove();
   });
 
+  it("releases stalled log hydration and permits a fresh retry without accepting late data", async () => {
+    vi.useFakeTimers();
+    const container = document.createElement("div");
+    const root = createRoot(container);
+    let latest!: ReturnType<typeof useLiveRunTranscripts>;
+    let resolveLate!: (result: Awaited<ReturnType<typeof logMock>>) => void;
+    const empty = { runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 0 };
+    logMock.mockImplementationOnce(() => new Promise((resolve) => { resolveLate = resolve; }));
+    function Harness() {
+      latest = useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "succeeded", adapterType: "codex_local" }] });
+      return null;
+    }
+    try {
+      await act(async () => { root.render(<Harness />); });
+      expect(latest.isInitialHydrating).toBe(true);
+      await act(async () => { await vi.advanceTimersByTimeAsync(TRANSCRIPT_REQUEST_TIMEOUT_MS); });
+      expect(latest.isInitialHydrating).toBe(false);
+      expect(latest.errorsByRun.get("run-1")?.message).toContain("too long");
+      await act(async () => { resolveLate(empty); });
+      expect(latest.errorsByRun.has("run-1")).toBe(true);
+      logMock.mockResolvedValue(empty);
+      await act(async () => { latest.retry(); });
+      expect(latest.errorsByRun.size).toBe(0);
+      expect(logMock).toHaveBeenCalledTimes(2);
+    } finally {
+      act(() => root.unmount());
+      vi.useRealTimers();
+    }
+  });
+
   it("stops retrying terminal runs whose persisted log never existed", async () => {
     logMock.mockReset();
     logMock.mockRejectedValue(new ApiError("Run log not found", 404, { error: "Run log not found" }));
@@ -261,7 +292,7 @@ describe("useLiveRunTranscripts", () => {
     });
 
     expect(logMock).toHaveBeenCalledTimes(1);
-    expect(logMock).toHaveBeenCalledWith("run-queued", 0, 256_000);
+    expect(logMock).toHaveBeenCalledWith("run-queued", 0, 256_000, expect.objectContaining({ signal: expect.any(AbortSignal) }));
     expect(FakeWebSocket.instances).toHaveLength(1);
 
     act(() => {
@@ -291,7 +322,7 @@ describe("useLiveRunTranscripts", () => {
     });
 
     expect(FakeWebSocket.instances).toHaveLength(0);
-    expect(logMock).toHaveBeenCalledWith("run-1", 0, 64_000);
+    expect(logMock).toHaveBeenCalledWith("run-1", 0, 64_000, expect.objectContaining({ signal: expect.any(AbortSignal) }));
 
     act(() => {
       root.unmount();
@@ -319,7 +350,7 @@ describe("useLiveRunTranscripts", () => {
       await Promise.resolve();
     });
 
-    expect(logMock).toHaveBeenCalledWith("run-1", 36_000, 64_000);
+    expect(logMock).toHaveBeenCalledWith("run-1", 36_000, 64_000, expect.objectContaining({ signal: expect.any(AbortSignal) }));
 
     act(() => {
       root.unmount();

ui/src/components/transcript/useLiveRunTranscripts.ts

@@ -1,4 +1,5 @@
-import { useEffect, useMemo, useRef, useState } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
+import { readTranscriptRequest } from "./read-transcript-request";
 import { useQuery } from "@tanstack/react-query";
 import type { LiveEvent } from "@paperclipai/shared";
 import { ApiError } from "../../api/client";
@@ -123,6 +124,12 @@ export function useLiveRunTranscripts({
   const normalizedRuns = useMemo(() => runs.map((run) => ({ ...run })), [runsKey]);
   const [chunksByRun, setChunksByRun] = useState<Map<string, RunLogChunk[]>>(new Map());
   const [hydratedRunIds, setHydratedRunIds] = useState<Set<string>>(new Set());
+  const [errorsByRun, setErrorsByRun] = useState<ReadonlyMap<string, Error>>(new Map());
+  const [retryGeneration, setRetryGeneration] = useState(0);
+  const retry = useCallback(() => {
+    missingTerminalLogRunIdsRef.current.clear();
+    setRetryGeneration((value) => value + 1);
+  }, []);
   const seenChunkKeysRef = useRef(new Set<string>());
   // Highest sequenced chunk trimmed out of a run's retained window; older
   // records re-delivered by the other transport are dropped instead of being
@@ -242,6 +249,11 @@ export function useLiveRunTranscripts({
       return next.size === prev.size ? prev : next;
     });
 
+    setErrorsByRun((previous) => {
+      const next = new Map([...previous].filter(([id]) => retainedRunIds.has(id)));
+      return next.size === previous.size ? previous : next;
+    });
+
     for (const key of pendingLogRowsByRunRef.current.keys()) {
       const runId = key.replace(/:records$/, "");
       if (!retainedRunIds.has(runId)) {
@@ -285,16 +297,28 @@ export function useLiveRunTranscripts({
     if (readableRuns.length === 0) return;
 
     let cancelled = false;
+    const controller = new AbortController();
+    const inFlightRunIds = new Set<string>();
 
     const readRunLog = async (run: RunTranscriptSource) => {
-      if (missingTerminalLogRunIdsRef.current.has(run.id)) {
+      if (missingTerminalLogRunIdsRef.current.has(run.id) || inFlightRunIds.has(run.id)) {
         return;
       }
+      inFlightRunIds.add(run.id);
       const offset = logOffsetByRunRef.current.get(run.id) ?? resolveInitialLogOffset(run, logReadLimitBytes);
       try {
-        const result = await heartbeatsApi.log(run.id, offset, logReadLimitBytes);
+        const result = await readTranscriptRequest(
+          (signal) => heartbeatsApi.log(run.id, offset, logReadLimitBytes, { signal }),
+          controller.signal,
+        );
         if (cancelled) return;
 
+        setErrorsByRun((previous) => {
+          if (!previous.has(run.id)) return previous;
+          const next = new Map(previous);
+          next.delete(run.id);
+          return next;
+        });
         appendChunks(run.id, parsePersistedLogContent(run.id, result.content, pendingLogRowsByRunRef.current));
 
         if (result.nextOffset !== undefined) {
@@ -305,10 +329,26 @@ export function useLiveRunTranscripts({
           logOffsetByRunRef.current.set(run.id, offset + result.content.length);
         }
       } catch (error) {
-        if (error instanceof ApiError && error.status === 404 && isTerminalStatus(run.status)) {
-          missingTerminalLogRunIdsRef.current.add(run.id);
+        if (cancelled) return;
+        if (error instanceof ApiError && error.status === 404) {
+          setErrorsByRun((previous) => {
+            if (!previous.has(run.id)) return previous;
+            const next = new Map(previous);
+            next.delete(run.id);
+            return next;
+          });
+          // A newly started run may not have created its log yet.
+          if (isTerminalStatus(run.status)) missingTerminalLogRunIdsRef.current.add(run.id);
+        } else {
+          setErrorsByRun((previous) => {
+            if (previous.has(run.id)) return previous;
+            const next = new Map(previous);
+            next.set(run.id, error instanceof Error ? error : new Error("Run history could not be loaded"));
+            return next;
+          });
         }
       } finally {
+        inFlightRunIds.delete(run.id);
         if (!cancelled) {
           setHydratedRunIds((prev) => {
             if (prev.has(run.id)) return prev;
@@ -340,9 +380,10 @@ export function useLiveRunTranscripts({
 
     return () => {
       cancelled = true;
+      controller.abort();
       if (interval !== null) window.clearInterval(interval);
     };
-  }, [enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey]);
+  }, [enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey, retryGeneration]);
 
   useEffect(() => {
     if (!enableRealtimeUpdates) return;
@@ -519,6 +560,7 @@ export function useLiveRunTranscripts({
 
   return {
     transcriptByRun,
+    hydratedRunIds, errorsByRun, retry,
     isInitialHydrating: normalizedRuns.some((run) => canReadPersistedLog(run) && !hydratedRunIds.has(run.id)),
     hasOutputForRun(runId: string) {
       return (chunksByRun.get(runId)?.length ?? 0) > 0 || runById.get(runId)?.hasStoredOutput === true;

ui/src/components/transcript/useNativeRunTranscripts.test.tsx

@@ -4,6 +4,7 @@ import { act } from "react";
 import { createRoot, type Root } from "react-dom/client";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { useNativeRunTranscripts } from "./useNativeRunTranscripts";
+import { TRANSCRIPT_REQUEST_TIMEOUT_MS } from "./read-transcript-request";
 
 const eventsMock = vi.hoisted(() => vi.fn());
 
@@ -86,3 +87,83 @@ describe("useNativeRunTranscripts", () => {
     expect(eventsMock.mock.calls.at(-1)?.[0]).toBe("failed-run");
   });
 });
+
+// Initial readiness is separate from an empty transcript: the task shell must
+// not reveal empty history while the first durable page is still in flight.
+describe("native history readiness and stable projection", () => {
+  let container: HTMLDivElement;
+  let root: Root;
+  let latest: ReturnType<typeof useNativeRunTranscripts>;
+  function StateProbe({ runs = [{ id: "one", status: "running", runtimeMode: "native" as const }] }) {
+    latest = useNativeRunTranscripts(runs);
+    return null;
+  }
+  beforeEach(() => {
+    vi.useFakeTimers();
+    eventsMock.mockReset();
+    container = document.createElement("div");
+    root = createRoot(container);
+  });
+  afterEach(() => {
+    act(() => root.unmount());
+    vi.useRealTimers();
+  });
+
+  it("distinguishes pending, loaded-empty and failed, and supports explicit retry", async () => {
+    let resolve!: (rows: never[]) => void;
+    eventsMock.mockImplementationOnce(() => new Promise((done) => { resolve = done; }));
+    await act(async () => { root.render(<StateProbe />); });
+    expect(latest.isInitialHydrating).toBe(true);
+    await act(async () => { resolve([]); });
+    expect(latest.isInitialHydrating).toBe(false);
+    expect(latest.hydratedRunIds.has("one")).toBe(true);
+    eventsMock.mockRejectedValueOnce(new Error("offline"));
+    await act(async () => { await vi.advanceTimersByTimeAsync(2_000); });
+    expect(latest.errorsByRun.has("one")).toBe(true);
+    expect(latest.isInitialHydrating).toBe(false);
+    eventsMock.mockResolvedValue([]);
+    await act(async () => { latest.retry(); });
+    expect(latest.errorsByRun.size).toBe(0);
+  });
+
+  it("does not rebuild unchanged history on empty polls", async () => {
+    eventsMock.mockResolvedValueOnce([{ seq: 1, payload: {}, eventType: "log" }]).mockResolvedValue([]);
+    await act(async () => { root.render(<StateProbe />); });
+    const projection = latest.transcriptByRun;
+    const rows = projection.get("one");
+    await act(async () => { await vi.advanceTimersByTimeAsync(2_000); });
+    expect(latest.transcriptByRun).toBe(projection);
+    expect(latest.transcriptByRun.get("one")).toBe(rows);
+  });
+
+  it("times out stalled history, aborts its request, and ignores a late response before retry", async () => {
+    let resolveLate!: (rows: never[]) => void;
+    eventsMock.mockImplementationOnce(() => new Promise((resolve) => { resolveLate = resolve; }));
+    await act(async () => { root.render(<StateProbe />); });
+    const signal = eventsMock.mock.calls[0][3].signal as AbortSignal;
+    expect(latest.isInitialHydrating).toBe(true);
+    await act(async () => { await vi.advanceTimersByTimeAsync(TRANSCRIPT_REQUEST_TIMEOUT_MS); });
+    expect(signal.aborted).toBe(true);
+    expect(latest.isInitialHydrating).toBe(false);
+    expect(latest.errorsByRun.get("one")?.message).toContain("too long");
+    await act(async () => { resolveLate([]); });
+    expect(latest.errorsByRun.has("one")).toBe(true);
+    eventsMock.mockResolvedValue([]);
+    await act(async () => { latest.retry(); });
+    expect(latest.errorsByRun.size).toBe(0);
+    expect(eventsMock.mock.calls[1][3].signal.aborted).toBe(false);
+  });
+
+  it("commits a resolved run independently of a stalled sibling and retains its cursor", async () => {
+    const rows = [{ seq: 7, payload: {}, eventType: "log" }];
+    eventsMock.mockImplementation((id) => id === "slow" ? new Promise(() => {}) : Promise.resolve(rows));
+    const one = { id: "one", status: "succeeded", runtimeMode: "native" as const };
+    await act(async () => { root.render(<StateProbe runs={[one, { ...one, id: "slow" }]} />); });
+    expect(latest.hydratedRunIds.has("one")).toBe(true);
+    expect(latest.hydratedRunIds.has("slow")).toBe(false);
+    expect(latest.transcriptByRun.has("one")).toBe(true);
+    await act(async () => { root.render(<StateProbe runs={[one]} />); });
+    expect(eventsMock.mock.calls.filter(([id]) => id === "one").map(([, cursor]) => cursor)).toEqual([0, 7]);
+    expect(latest.isInitialHydrating).toBe(false);
+  });
+});

ui/src/components/transcript/useNativeRunTranscripts.ts

@@ -1,8 +1,9 @@
-import { useEffect, useMemo, useRef, useState } from "react";
+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;
@@ -36,90 +37,116 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource
   );
   const [eventsByRun, setEventsByRun] = useState<Map<string, HeartbeatRunEvent[]>>(new Map());
   const [errorsByRun, setErrorsByRun] = useState<Map<string, NativeRunTranscriptError>>(new Map());
+  const [hydratedRunIds, setHydratedRunIds] = useState<ReadonlySet<string>>(new Set());
+  const [retryGeneration, setRetryGeneration] = useState(0);
+  const retry = useCallback(() => setRetryGeneration((value) => value + 1), []);
+  const projectionCacheRef = useRef(new Map<string, { events: HeartbeatRunEvent[]; transcript: TranscriptEntry[] }>());
   const cursorByRunRef = useRef(new Map<string, number>());
 
   useEffect(() => {
     let cancelled = false;
-    let timer: number | null = null;
+    const controller = new AbortController();
+    const timers = new Set<number>();
+    const retainedIds = new Set(nativeRuns.map((run) => run.id));
+    const retainMap = <T,>(previous: Map<string, T>) => {
+      const next = new Map([...previous].filter(([id]) => retainedIds.has(id)));
+      return next.size === previous.size ? previous : next;
+    };
+    setEventsByRun(retainMap);
+    setErrorsByRun(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 refresh = async (runsToRefresh: readonly NativeRunTranscriptSource[]) => {
-      const updates = new Map<string, HeartbeatRunEvent[]>();
-      const errors = new Map<string, NativeRunTranscriptError>();
-      await Promise.all(runsToRefresh.map(async (run) => {
-        try {
-          let cursor = cursorByRunRef.current.get(run.id) ?? 0;
-          const incoming: HeartbeatRunEvent[] = [];
-          for (;;) {
-            const page = await heartbeatsApi.events(run.id, cursor, EVENT_PAGE_SIZE);
-            if (cancelled) return;
-            const last = page.at(-1);
-            const nextCursor = last ? Math.max(cursor, last.seq) : cursor;
-            incoming.push(...page);
-            if (page.length < EVENT_PAGE_SIZE || nextCursor === cursor) {
-              cursor = nextCursor;
-              break;
-            }
+    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;
           }
-          if (incoming.length > 0) updates.set(run.id, incoming);
-          cursorByRunRef.current.set(run.id, cursor);
-        } catch (error) {
-          // Keep the last durable cursor; the next poll retries this run only.
-          errors.set(run.id, {
+          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;
+        });
+      } 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(),
           });
-        }
-      }));
-
-      if (cancelled) return;
-      const retainedIds = new Set(nativeRuns.map((run) => run.id));
-      for (const runId of cursorByRunRef.current.keys()) {
-        if (!retainedIds.has(runId)) cursorByRunRef.current.delete(runId);
+          return next;
+        });
       }
-      setEventsByRun((previous) => {
-        const next = new Map<string, HeartbeatRunEvent[]>();
-        for (const runId of retainedIds) {
-          const current = previous.get(runId) ?? [];
-          const incoming = updates.get(runId) ?? [];
-          next.set(runId, incoming.length > 0 ? [...current, ...incoming] : current);
-        }
-        return next;
-      });
-      setErrorsByRun((previous) => {
-        const next = new Map<string, NativeRunTranscriptError>();
-        for (const runId of retainedIds) {
-          const error = errors.get(runId);
-          if (error) next.set(runId, previous.get(runId) ?? error);
-        }
-        return next;
-      });
-
-      if (nativeRuns.some((run) => isLive(run.status)) || errors.size > 0) {
-        const retryRuns = nativeRuns.filter(
-          (run) => isLive(run.status) || errors.has(run.id),
-        );
-        timer = window.setTimeout(
-          () => void refresh(retryRuns),
-          EVENT_POLL_INTERVAL_MS,
-        );
+      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);
       }
     };
-
-    void refresh(nativeRuns);
+    for (const run of nativeRuns) void refreshRun(run);
     return () => {
       cancelled = true;
-      if (timer !== null) window.clearTimeout(timer);
+      controller.abort();
+      for (const timer of timers) window.clearTimeout(timer);
     };
-  }, [nativeRuns]);
+  }, [nativeRuns, retryGeneration]);
 
   const transcriptByRun = useMemo(() => {
     const transcripts = new Map<string, TranscriptEntry[]>();
     for (const run of nativeRuns) {
-      transcripts.set(run.id, nativeRunEventsToTranscript(eventsByRun.get(run.id) ?? []));
+      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]);
 
-  return { transcriptByRun, errorsByRun };
+  return {
+    transcriptByRun, errorsByRun, hydratedRunIds, retry,
+    isInitialHydrating: nativeRuns.some((run) => !hydratedRunIds.has(run.id)),
+  };
 }