diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 0e6f9fdf2f9..d0a5993faa9 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -532,8 +532,11 @@ retain the existing refresh fallback. Run transitions, activity events, and reconnect recovery still refresh their affected data. Task run queries stop interval polling only after a successful empty response -and an open shared live connection for the current company and session. Initial -reads, active work, errors, connection setup, and disconnections retain their +and an open shared live connection for the current company and session. With +that connection open, a successful active response (a run list or one run) +polls only every 15 seconds. Lifecycle and progress events keep it current; the +slow read only bounds drift in fields that have no event. Initial reads, +unrecognized data, errors, connection setup, and disconnections retain their existing polling cadence. Company or session changes require a new connection. Views without an assigned agent retain polling. Matching run lifecycle events refresh the displayed task's run queries independently of route and window @@ -546,6 +549,18 @@ queries after any older request. Follow-up reads belong to the current mounted scope and stop when it changes or unmounts. The first open and reconnect still reconcile missed events. +Live events refresh company-wide lists and projections through a slower +invalidation lane. These are task lists and summaries, the agent list, run +lists, badges, the dashboard, costs and the company activity feed. The lane +flushes at most once every 5 seconds. After 5 quiet seconds, an event refreshes +them after the normal batch interval. An event inside the 5-second window waits +until the window ends. Task detail, comments, runs, browsers and other +per-entity queries keep the fast lane. A task's browser list stops interval +polling while the live connection is open and no browser can still change +state. Browser lifecycle activity on that task refreshes it. If a browser read +is in flight, one shared follow-up read runs after it settles. A list that no +view observes becomes stale, so the next view reads it. + Bundled skill release checks read the current stored version ID and a SHA-256 fingerprint of its file inventory. A per-database memo keeps at most 64 verified snapshots. It reuses a verification only when the current database fingerprint diff --git a/ui/src/components/task-side-panel/TaskSidePanel.tsx b/ui/src/components/task-side-panel/TaskSidePanel.tsx index 05b37c3026e..e04a616e91f 100644 --- a/ui/src/components/task-side-panel/TaskSidePanel.tsx +++ b/ui/src/components/task-side-panel/TaskSidePanel.tsx @@ -393,7 +393,7 @@ function TaskSidePanelForScope({ controller.openTab(taskPanelPluginTab(panel), firstArrival); } }, [dismissedPluginPanels, openedPluginPanels, pluginPanelKey, pluginPanels, pluginPanelsLoading, controller.tabs, controller.closeTab, controller.openTab]); - const browsersQuery = useTaskBrowsers(issue.id); + const browsersQuery = useTaskBrowsers(issue.id, issue.companyId); const newestLiveBrowser = browsersQuery.data?.findLast((browser) => browser.status === "running" || browser.status === "idle"); useEffect(() => { if (!openBrowserId) return; diff --git a/ui/src/context/LiveUpdatesProvider.fanout.test.ts b/ui/src/context/LiveUpdatesProvider.fanout.test.ts new file mode 100644 index 00000000000..86c74155784 --- /dev/null +++ b/ui/src/context/LiveUpdatesProvider.fanout.test.ts @@ -0,0 +1,206 @@ +// @vitest-environment jsdom + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { QueryClient, QueryObserver, type QueryObserverOptions } from "@tanstack/react-query"; +import type { TaskBrowser } from "@paperclipai/shared"; +import { __liveUpdatesTestUtils } from "./LiveUpdatesProvider"; +import { createCoalescingQueryClient } from "../lib/query-invalidation-batcher"; +import { liveRunRefetchInterval } from "../lib/live-runs-cache"; +import { taskBrowsersQueryKey, taskBrowsersRefetchInterval } from "../hooks/useTaskBrowsers"; +import { queryKeys } from "../lib/queryKeys"; + +const COMPANY = "company-1"; +const TASK_UUID = "11111111-1111-4111-8111-111111111111"; +const TASK_REF = "DEMO-1"; +const ACTOR = { userId: "viewer", agentId: null }; +// The staging capture that motivated this test observed one task for 17 seconds. +const WINDOW_MS = 17_000; + +type Counter = ReturnType; + +const liveRun = { + id: "run-open", status: "running", invocationSource: "assignment", triggerDetail: null, + startedAt: "2026-11-09T00:00:00.000Z", finishedAt: null, createdAt: "2026-11-09T00:00:00.000Z", + agentId: "agent-1", agentName: "Agent One", adapterType: "claude_local", issueId: TASK_UUID, +}; + +function observe(client: QueryClient, options: QueryObserverOptions) { + const observer = new QueryObserver(client, options); + return observer.subscribe(() => {}); +} + +/** + * Mount the reads an open, running task keeps active: the sidebar's recent-task + * summaries and agent list, the company run list, the task's child lists, its + * run list (two observers at the task view's two intervals), and its browsers. + */ +function mountOpenTask(client: QueryClient, connected: boolean) { + const counters: Record = {}; + const read = (name: string, data: T) => { + counters[name] ??= vi.fn(async () => data); + return counters[name] as unknown as () => Promise; + }; + const stale = { staleTime: 30_000 }; + const unsubscribes = [ + observe(client, { queryKey: queryKeys.issues.summaryBatch(COMPANY, "viewer", ["a", "b"]), queryFn: read("summaries", []), ...stale }), + observe(client, { queryKey: queryKeys.agents.list(COMPANY), queryFn: read("agents", []), ...stale }), + observe(client, { queryKey: queryKeys.liveRuns(COMPANY), queryFn: read("companyLiveRuns", [liveRun]), ...stale }), + observe(client, { queryKey: queryKeys.issues.listByParent(COMPANY, TASK_UUID), queryFn: read("parentList", []), ...stale }), + observe(client, { queryKey: queryKeys.issues.listCreatedFromIssue(COMPANY, TASK_UUID), queryFn: read("createdFromList", []), ...stale }), + observe(client, { queryKey: queryKeys.issues.listByDescendantRoot(COMPANY, TASK_UUID), queryFn: read("descendantList", []), ...stale }), + ...[1000, 3000].map((fallback) => observe(client, { + queryKey: queryKeys.issues.liveRuns(TASK_REF), queryFn: read("taskLiveRuns", [liveRun]), ...stale, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, fallback), + })), + observe(client, { + queryKey: taskBrowsersQueryKey(TASK_UUID), queryFn: read("taskBrowsers", [] as TaskBrowser[]), ...stale, + refetchInterval: (query) => taskBrowsersRefetchInterval(connected, query.state), + }), + ]; + return { counters, unmount: () => unsubscribes.forEach((unsubscribe) => unsubscribe()) }; +} + +/** + * A busy company: other agents' runs change state about once a second and + * other tasks log activity a little more often. None of it is about the open task. + */ +function deliverBusyCompanyStream(client: QueryClient) { + const gate = { cooldownHits: new Map(), suppressUntil: 0, observedRunOutcomes: new Set() }; + let id = 0; + const deliver = (type: string, payload: Record) => + __liveUpdatesTestUtils.handleLiveEvent(client, COMPANY, `/DEMO/issues/${TASK_REF}`, { + id: ++id, companyId: COMPANY, type, createdAt: new Date().toISOString(), payload, + } as Parameters[3], () => null, gate, ACTOR); + const timers = [ + setInterval(() => deliver("heartbeat.run.status", { + runId: `run-other-${id % 4}`, agentId: `agent-${id % 4}`, issueId: `other-${id % 4}`, status: "running", + }), 1000), + setInterval(() => deliver("activity.logged", { + entityType: "issue", entityId: `other-${id % 5}`, action: "issue.comment_added", actorType: "agent", actorId: "agent-2", + }), 700), + ]; + return () => timers.forEach((timer) => clearInterval(timer)); +} + +async function countRequests(connected: boolean, wire: (client: QueryClient) => { client: QueryClient; dispose: () => void }) { + const client = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + const live = wire(client); + const task = mountOpenTask(client, connected); + await vi.advanceTimersByTimeAsync(0); + const stop = connected ? deliverBusyCompanyStream(live.client) : () => {}; + await vi.advanceTimersByTimeAsync(WINDOW_MS); + stop(); + task.unmount(); + live.dispose(); + client.clear(); + return Object.fromEntries(Object.entries(task.counters).map(([name, fn]) => [name, fn.mock.calls.length])); +} + +const liveWiring = (client: QueryClient) => { + const batcher = __liveUpdatesTestUtils.createLiveInvalidationBatcher(client, COMPANY); + return { client: createCoalescingQueryClient(client, batcher), dispose: batcher.dispose }; +}; + +describe("open-task request fan-out", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-11-09T12:00:00.000Z")); + }); + afterEach(() => vi.useRealTimers()); + + it("bounds company-wide refetches and connected run and browser polling during a busy 17-second visit", async () => { + const counts = await countRequests(true, liveWiring); + // Printed for the before/after comparison in the pull request. + console.info("open-task fan-out (connected, busy company)", JSON.stringify(counts)); + // Ambient lists: the first mount plus at most one refresh per spacing window. + for (const name of ["summaries", "agents", "parentList", "createdFromList", "descendantList"]) { + expect(counts[name], name).toBeLessThanOrEqual(1 + Math.ceil(WINDOW_MS / 5_000)); + expect(counts[name], name).toBeGreaterThan(1); + } + // The task's run list waits for its own lifecycle events plus a slow safety read. + expect(counts.taskLiveRuns).toBeLessThanOrEqual(1 + Math.floor(WINDOW_MS / 15_000)); + // No browsers and an open connection: one read; browser activity refreshes it. + expect(counts.taskBrowsers).toBe(1); + }); + + it("keeps the polling fallback while the live connection is closed", async () => { + const counts = await countRequests(false, (client) => ({ client, dispose: () => {} })); + expect(counts.taskLiveRuns).toBeGreaterThanOrEqual(WINDOW_MS / 1000); + expect(counts.taskBrowsers).toBeGreaterThanOrEqual(WINDOW_MS / 3000); + }); + + it("refreshes ambient lists after one normal batch when the company is quiet", async () => { + const client = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + const live = liveWiring(client); + const task = mountOpenTask(client, true); + await vi.advanceTimersByTimeAsync(0); + __liveUpdatesTestUtils.invalidateActivityQueries(live.client, COMPANY, + { entityType: "issue", entityId: "other-1", action: "issue.updated", actorType: "agent", actorId: "agent-2" }, ACTOR); + await vi.advanceTimersByTimeAsync(400); + expect(task.counters.summaries).toHaveBeenCalledTimes(2); + expect(task.counters.descendantList).toHaveBeenCalledTimes(2); + task.unmount(); + live.dispose(); + client.clear(); + }); +}); + +describe("task browser refresh", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("refreshes the open task's browsers on its browser activity only", async () => { + const client = new QueryClient({ defaultOptions: { queries: { retry: false } } }); + const live = liveWiring(client); + const task = mountOpenTask(client, true); + await vi.advanceTimersByTimeAsync(10_000); + expect(task.counters.taskBrowsers).toHaveBeenCalledTimes(1); + const activity = (entityId: string, action: string) => __liveUpdatesTestUtils.invalidateActivityQueries(live.client, COMPANY, + { entityType: "issue", entityId, action, actorType: "system", actorId: "browser-use-cloud" }, ACTOR); + activity(TASK_UUID, "browser.started"); + await vi.advanceTimersByTimeAsync(400); + expect(task.counters.taskBrowsers).toHaveBeenCalledTimes(2); + activity("other-1", "browser.started"); + activity(TASK_UUID, "issue.comment_added"); + await vi.advanceTimersByTimeAsync(10_000); + expect(task.counters.taskBrowsers).toHaveBeenCalledTimes(2); + task.unmount(); + live.dispose(); + client.clear(); + }); +}); + +describe("ambient invalidation lane", () => { + const { isAmbientCompanyQueryKey } = __liveUpdatesTestUtils; + + it.each([ + queryKeys.issues.list(COMPANY), + queryKeys.issues.summaryBatch(COMPANY, "viewer", ["a"]), + queryKeys.issues.listByDescendantRoot(COMPANY, TASK_UUID), + queryKeys.agents.list(COMPANY), + queryKeys.liveRuns(COMPANY), + queryKeys.heartbeats(COMPANY), + queryKeys.sidebarBadges(COMPANY), + queryKeys.dashboard(COMPANY), + queryKeys.activity(COMPANY), + ].map((key) => [key]))("spaces the company-wide key %j", (queryKey) => { + expect(isAmbientCompanyQueryKey(COMPANY, queryKey)).toBe(true); + }); + + it.each([ + queryKeys.issues.detail(TASK_REF), + queryKeys.issues.comments(TASK_UUID), + queryKeys.issues.liveRuns(TASK_REF), + queryKeys.issues.activeRun(TASK_UUID), + queryKeys.agents.detail("agent-1"), + taskBrowsersQueryKey(TASK_UUID), + queryKeys.attention(COMPANY), + queryKeys.issues.list("company-2"), + ].map((key) => [key]))("keeps %j on the fast lane", (queryKey) => { + expect(isAmbientCompanyQueryKey(COMPANY, queryKey)).toBe(false); + }); + + it("uses no ambient lane without a live company", () => { + expect(isAmbientCompanyQueryKey(null, queryKeys.issues.list(COMPANY))).toBe(false); + }); +}); diff --git a/ui/src/context/LiveUpdatesProvider.recovery.test.tsx b/ui/src/context/LiveUpdatesProvider.recovery.test.tsx index 2f7f9d8813d..139cedcb6e8 100644 --- a/ui/src/context/LiveUpdatesProvider.recovery.test.tsx +++ b/ui/src/context/LiveUpdatesProvider.recovery.test.tsx @@ -17,7 +17,9 @@ vi.mock("../api/auth", () => ({ })); vi.mock("../api/health", () => ({ healthApi: { get: healthRead } })); import { LiveUpdatesProvider, useHasCompanyLiveConnection, useTaskRunLiveUpdates } from "./LiveUpdatesProvider"; -import { liveRunRefetchInterval } from "../lib/live-runs-cache"; +import { CONNECTED_ACTIVE_RUN_REFETCH_MS, liveRunRefetchInterval } from "../lib/live-runs-cache"; +import { taskBrowsersQueryKey } from "../lib/task-browsers-refresh"; +import { taskBrowsersRefetchInterval } from "../hooks/useTaskBrowsers"; (globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; @@ -184,11 +186,88 @@ describe("LiveUpdatesProvider connection recovery", () => { expect(pollRead).toHaveBeenCalledTimes(readsAfterOpen); }); - it.each([[{ id: "run-1" }], { id: "run-1" }])("keeps active work polling while connected for %j", async (data) => { + it.each([ + ["a run list", [{ id: "run-1", status: "running" }]], + ["a single run", { id: "run-1", status: "running" }], + ])("slows active work polling to the safety interval while connected for %s", async (_label, data) => { vi.stubGlobal("WebSocket", Socket); pollRead.mockResolvedValue(data); await renderPolling(); await act(async () => Socket.instances[0].onopen?.()); + await advancePolls(1); + pollRead.mockClear(); + // Lifecycle events refresh active runs while connected; no one-second reads. + await advancePolls(5); + expect(pollRead).not.toHaveBeenCalled(); + await advancePolls(Math.ceil(CONNECTED_ACTIVE_RUN_REFETCH_MS / 1000)); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(1); + }); + + it("reads task browsers again when the connection opens during an older empty read", async () => { + vi.stubGlobal("WebSocket", Socket); + const started = [{ id: "browser-1", status: "running" }]; + let release!: (value: unknown) => void; + const browsersRead = vi.fn<() => Promise>().mockResolvedValue(started); + function Browsers() { + const connected = useHasCompanyLiveConnection("company-1"); + useQuery({ + queryKey: taskBrowsersQueryKey("task-1"), queryFn: browsersRead, + refetchInterval: (query) => taskBrowsersRefetchInterval(connected, query.state as Parameters[1]), + }); + return null; + } + browsersRead.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + await act(async () => root.render( + , + )); + await act(async () => vi.advanceTimersByTimeAsync(1)); + expect(browsersRead).toHaveBeenCalledTimes(1); + // The browser starts while the socket is closed; its activity is not replayed. + await act(async () => Socket.instances[0].onopen?.()); + await act(async () => release([])); + await act(async () => vi.advanceTimersByTimeAsync(1)); + expect(browsersRead).toHaveBeenCalledTimes(2); + expect(client.getQueryData(taskBrowsersQueryKey("task-1"))).toEqual(started); + }); + + it("reads an idle task browser list once when the connection opens", async () => { + vi.stubGlobal("WebSocket", Socket); + const browsersRead = vi.fn<() => Promise>().mockResolvedValue([]); + function Browsers() { + const connected = useHasCompanyLiveConnection("company-1"); + useQuery({ + queryKey: taskBrowsersQueryKey("task-1"), queryFn: browsersRead, + refetchInterval: (query) => taskBrowsersRefetchInterval(connected, query.state as Parameters[1]), + }); + return null; + } + await act(async () => root.render( + , + )); + await act(async () => vi.advanceTimersByTimeAsync(1)); + browsersRead.mockClear(); + await act(async () => Socket.instances[0].onopen?.()); + await act(async () => vi.advanceTimersByTimeAsync(10_000)); + // The open reconcile reads it once; it is not followed up as an older read. + expect(browsersRead).toHaveBeenCalledTimes(1); + }); + + it("keeps one-second polling for active work after the connection closes", async () => { + vi.stubGlobal("WebSocket", Socket); + pollRead.mockResolvedValue([{ id: "run-1", status: "running" }]); + await renderPolling(); + await act(async () => Socket.instances[0].onopen?.()); + await act(async () => Socket.instances[0].onclose?.()); + pollRead.mockClear(); + await advancePolls(3); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(3); + }); + + it("keeps fallback polling for an unrecognized active result while connected", async () => { + vi.stubGlobal("WebSocket", Socket); + pollRead.mockResolvedValue({ id: "run-1" }); + await renderPolling(); + await act(async () => Socket.instances[0].onopen?.()); pollRead.mockClear(); await advancePolls(3); expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(3); diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index ed94ff641c6..86ca8a050c6 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -3,6 +3,7 @@ import { resolveAgentAppearance, TASK_IDENTITY_CHANGED_FIELD, issueActivityChang import { getPageVisibility, usePageVisibility } from "../lib/page-visibility"; import { createLiveQueryRecovery } from "../lib/live-query-recovery"; import { isLiveReconciledQuery } from "../lib/live-reconcile"; +import { refreshPendingTaskBrowserReads, refreshTaskBrowsers } from "../lib/task-browsers-refresh"; import { createContext, useCallback, @@ -21,9 +22,12 @@ import { type QueryClient, } from "@tanstack/react-query"; import { + AMBIENT_INVALIDATION_SPACING_MS, createCoalescingQueryClient, createInvalidationBatcher, + createLaneRoutingBatcher, excludingQueryKeys, + type InvalidationBatcher, } from "../lib/query-invalidation-batcher"; import { patchRunStatusInList, @@ -1579,6 +1583,8 @@ function invalidateActivityQueries( } if (entityType === "issue") { + // Browser lists stop polling while connected; this activity refreshes them. + if (entityId && action?.startsWith("browser.")) refreshTaskBrowsers(queryClient, entityId); const chatListKey = queryKeys.agentChats.list(companyId, currentActor.userId); const knownChat = entityId && queryClient.getQueryData(chatListKey)?.some(chat => chat.id === entityId); const ownChatDelivery = readString(details?.conversationUserId) === (currentActor.userId ?? "local-board"); @@ -2153,6 +2159,41 @@ function handleLiveEvent( } } +/** + * Company-scoped query families whose data is a company-wide list or + * projection: task lists and summaries, the agent list, run lists, badges, + * the dashboard, costs and the company activity feed. Many events in a busy + * company change them, and no single task view depends on them for its own + * state. Task detail, comments, runs and other per-entity keys are not listed. + */ +const AMBIENT_COMPANY_QUERY_ROOTS: ReadonlySet = new Set([ + "issues", + "agents", + "heartbeats", + "live-runs", + "dashboard", + "sidebar-badges", + "costs", + "activity", +]); + +function isAmbientCompanyQueryKey(companyId: string | null, queryKey: readonly unknown[]): boolean { + return !!companyId && queryKey[1] === companyId + && typeof queryKey[0] === "string" && AMBIENT_COMPANY_QUERY_ROOTS.has(queryKey[0]); +} + +function createLiveInvalidationBatcher( + queryClient: QueryClient, + companyId: string | null, + ambientSpacingMs: number = AMBIENT_INVALIDATION_SPACING_MS, +): InvalidationBatcher { + return createLaneRoutingBatcher( + createInvalidationBatcher(queryClient), + createInvalidationBatcher(queryClient, undefined, { minSpacingMs: ambientSpacingMs }), + (queryKey) => isAmbientCompanyQueryKey(companyId, queryKey), + ); +} + function resolveLiveCompanyId( selectedCompanyId: string | null, selectedCompanyLiveId: string | null, @@ -2196,6 +2237,8 @@ function closeSocketQuietly( } export const __liveUpdatesTestUtils = { + createLiveInvalidationBatcher, + isAmbientCompanyQueryKey, MENTION_POOL_REFRESH_WINDOW_MS, buildActivityToast, applyRunLifecycleToCompanyLiveRuns, @@ -2272,10 +2315,12 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { // Coalesce the per-event invalidation storm. Optimistic setQueryData writes // still pass straight through (immediate); only invalidateQueries is batched - // and flushed at most a few times per second. + // and flushed at most a few times per second. Company-wide lists and + // projections use a slower lane, so a busy company does not refetch them on + // every batch while a task is open. const invalidationBatcher = useMemo( - () => createInvalidationBatcher(queryClient), - [queryClient], + () => createLiveInvalidationBatcher(queryClient, liveCompanyId), + [queryClient, liveCompanyId], ); const coalescingClient = useMemo( () => createCoalescingQueryClient(queryClient, invalidationBatcher), @@ -2383,6 +2428,10 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { // too. Reconcile that gap as well as reconnects: events are not replayed. const initialHealthyOpen = !hasOpenedSocket.current && !hasNeededSocketRecovery.current; hasOpenedSocket.current = true; + // Browser lists stop polling on an empty result. A reconcile that joins + // an older read needs one more read after it. Scan before reconciling, + // so the reconcile's own reads are not followed up again. + refreshPendingTaskBrowserReads(queryClient); void queryClient.invalidateQueries({ type: "active", predicate: initialHealthyOpen ? (query) => { @@ -2403,6 +2452,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { const parsed = JSON.parse(raw) as LiveEvent; if (transcripts.receive(parsed, recoverQueries)) { if ((parsed as { type: string }).type === "live.resync") { + refreshPendingTaskBrowserReads(queryClient); void queryClient.invalidateQueries({ type: "active", predicate: isLiveReconciledQuery }, { cancelRefetch: false }); } return; diff --git a/ui/src/hooks/useTaskBrowsers.test.ts b/ui/src/hooks/useTaskBrowsers.test.ts index 23b8e0638d2..fe572bf4338 100644 --- a/ui/src/hooks/useTaskBrowsers.test.ts +++ b/ui/src/hooks/useTaskBrowsers.test.ts @@ -5,7 +5,7 @@ import { afterEach, beforeEach, expect, it, vi } from "vitest"; import type { TaskBrowser } from "@paperclipai/shared"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { browserUseApi } from "@/api/browser-use"; -import { useBrowserArrivals, useTaskBrowsers } from "./useTaskBrowsers"; +import { TASK_BROWSERS_REFETCH_MS, taskBrowsersRefetchInterval, useBrowserArrivals, useTaskBrowsers } from "./useTaskBrowsers"; (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; const cleanups = new Set<() => void>(); function cleanup() { for (const unmount of cleanups) unmount(); } @@ -105,3 +105,18 @@ it("does not steal focus when browser-tab storage is unavailable", () => { rerender(); expect(result.current.openBrowserId).toBe(null); }); +it.each([ + ["no browsers", [], false], + ["only finished browsers", [{ ...browser("a"), status: "closed" }, { ...browser("b"), status: "failed" }], false], + ["a starting browser", [{ ...browser("a"), status: "starting" }], TASK_BROWSERS_REFETCH_MS], + ["an idle browser", [browser("a")], TASK_BROWSERS_REFETCH_MS], + ["a stopping browser", [{ ...browser("a"), status: "stopping" }], TASK_BROWSERS_REFETCH_MS], +] as Array<[string, TaskBrowser[], number | false]>)("with the live connection open, polls %s at %s", (_label, data, expected) => { + expect(taskBrowsersRefetchInterval(true, { status: "success", data })).toBe(expected); + // A closed connection always keeps the polling fallback. + expect(taskBrowsersRefetchInterval(false, { status: "success", data })).toBe(TASK_BROWSERS_REFETCH_MS); +}); +it.each(["pending", "error"] as const)("keeps polling a %s browser read while connected", (status) => { + expect(taskBrowsersRefetchInterval(true, { status, data: [] })).toBe(TASK_BROWSERS_REFETCH_MS); + expect(taskBrowsersRefetchInterval(true, { status, data: undefined })).toBe(TASK_BROWSERS_REFETCH_MS); +}); diff --git a/ui/src/hooks/useTaskBrowsers.ts b/ui/src/hooks/useTaskBrowsers.ts index b9711fd8b86..9e85a2dc782 100644 --- a/ui/src/hooks/useTaskBrowsers.ts +++ b/ui/src/hooks/useTaskBrowsers.ts @@ -3,12 +3,39 @@ import { useCallback, useEffect, useRef, useState } from "react"; import type { TaskBrowser } from "@paperclipai/shared"; import { browserUseApi } from "@/api/browser-use"; import { retryTransientOnly } from "@/lib/query-client"; -export function useTaskBrowsers(issueId?: string) { +import { useHasCompanyLiveConnection } from "@/context/LiveUpdatesProvider"; +import { taskBrowsersQueryKey } from "@/lib/task-browsers-refresh"; +export const TASK_BROWSERS_REFETCH_MS = 3000; + +/** + * Poll while the company live connection is closed, and while any browser can + * still change state. With the connection open, every browser lifecycle step + * (`browser.started`, `browser.ready`, `browser.run_finished`, `browser.closed` + * and controls) logs task activity, and `LiveUpdatesProvider` refreshes the + * list on that activity (`refreshTaskBrowsers`). + */ +export function taskBrowsersRefetchInterval( + connected: boolean, + state: { status: "pending" | "error" | "success"; data: TaskBrowser[] | undefined }, +): number | false { + if (!connected || state.status !== "success" || !Array.isArray(state.data)) return TASK_BROWSERS_REFETCH_MS; + return state.data.some(browser => browser.status !== "closed" && browser.status !== "failed") + ? TASK_BROWSERS_REFETCH_MS + : false; +} + +export { taskBrowsersQueryKey }; + +export function useTaskBrowsers(issueId?: string, companyId?: string | null) { + // Socket open and reconnect reconcile active queries, including this one. + // Browser activity refreshes it through `refreshTaskBrowsers`. + const connected = useHasCompanyLiveConnection(companyId); + const enabled = Boolean(issueId) && !issueId?.startsWith("chat:"); return useQuery({ - queryKey: ["task-browsers", issueId], + queryKey: taskBrowsersQueryKey(issueId), queryFn: () => browserUseApi.list(issueId!), - enabled: Boolean(issueId) && !issueId?.startsWith("chat:"), - refetchInterval: 3000, + enabled, + refetchInterval: (query) => taskBrowsersRefetchInterval(connected, query.state), // The 3s poll cadence already retries reads; stacking the shared backoff // inside a tick only adds duplicate requests. The next tick refetches anyway. diff --git a/ui/src/lib/live-runs-cache.test.ts b/ui/src/lib/live-runs-cache.test.ts index c908e44c590..a20c59309c3 100644 --- a/ui/src/lib/live-runs-cache.test.ts +++ b/ui/src/lib/live-runs-cache.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import type { LiveRunForIssue } from "../api/heartbeats"; -import { liveRunRefetchInterval, patchRunStatusInList, removeRunFromList } from "./live-runs-cache"; +import { CONNECTED_ACTIVE_RUN_REFETCH_MS, liveRunRefetchInterval, patchRunStatusInList, removeRunFromList } from "./live-runs-cache"; function run(id: string, status: string): LiveRunForIssue { return { @@ -73,8 +73,19 @@ describe("liveRunRefetchInterval", () => { expect(liveRunRefetchInterval(false, { status: "success", data }, 1000)).toBe(1000); }); - it.each([undefined, {}, [run("active", "running")], "invalid", 0])("preserves polling for unknown or active data %j", (data) => { + it.each([undefined, {}, { id: "run-1" }, "invalid", 0])("preserves polling for unknown data %j", (data) => { expect(liveRunRefetchInterval(true, { status: "success", data }, 3000)).toBe(3000); expect(liveRunRefetchInterval(false, { status: "success", data }, 3000)).toBe(3000); }); + + it.each([ + ["a run list", [run("active", "running")]], + ["a single run", run("active", "running")], + ])("waits for events with a slow safety read for %s while connected", (_label, data) => { + expect(liveRunRefetchInterval(true, { status: "success", data }, 1000)).toBe(CONNECTED_ACTIVE_RUN_REFETCH_MS); + expect(liveRunRefetchInterval(true, { status: "success", data }, 3000)).toBe(CONNECTED_ACTIVE_RUN_REFETCH_MS); + expect(liveRunRefetchInterval(true, { status: "success", data }, 60_000)).toBe(60_000); + // A closed connection keeps the caller's cadence. + expect(liveRunRefetchInterval(false, { status: "success", data }, 1000)).toBe(1000); + }); }); diff --git a/ui/src/lib/live-runs-cache.ts b/ui/src/lib/live-runs-cache.ts index 77b09ba29de..d0547ce882a 100644 --- a/ui/src/lib/live-runs-cache.ts +++ b/ui/src/lib/live-runs-cache.ts @@ -48,12 +48,29 @@ export function patchRunStatusInList( return { next: changed ? next : runs, present }; } +/** + * Safety interval for a task's known runs while the company live connection is + * open. Run lifecycle events refresh these queries (`useTaskRunLiveUpdates`) + * and progress events patch them in place, so the interval only bounds drift + * in server-derived fields that have no event, such as output silence. + */ +export const CONNECTED_ACTIVE_RUN_REFETCH_MS = 15_000; + export function liveRunRefetchInterval( connected: boolean, state: { status: "pending" | "error" | "success"; data: unknown }, fallbackMs: number, ): number | false { - return connected && state.status === "success" && (state.data === null || (Array.isArray(state.data) && state.data.length === 0)) - ? false - : fallbackMs; + if (!connected || state.status !== "success") return fallbackMs; + if (state.data === null || (Array.isArray(state.data) && state.data.length === 0)) return false; + // A known active result (a run list or a single run) waits for events. + if (Array.isArray(state.data) || isRunRecord(state.data)) { + return Math.max(fallbackMs, CONNECTED_ACTIVE_RUN_REFETCH_MS); + } + return fallbackMs; +} + +function isRunRecord(value: unknown): boolean { + return typeof value === "object" && value !== null && typeof (value as { id?: unknown }).id === "string" + && typeof (value as { status?: unknown }).status === "string"; } diff --git a/ui/src/lib/query-invalidation-batcher.test.ts b/ui/src/lib/query-invalidation-batcher.test.ts index a5c566f926f..7bb2cba9c21 100644 --- a/ui/src/lib/query-invalidation-batcher.test.ts +++ b/ui/src/lib/query-invalidation-batcher.test.ts @@ -3,6 +3,7 @@ import { QueryClient, QueryObserver } from "@tanstack/react-query"; import { createCoalescingQueryClient, createInvalidationBatcher, + createLaneRoutingBatcher, excludingQueryKeys, } from "./query-invalidation-batcher"; @@ -120,6 +121,77 @@ describe("createInvalidationBatcher", () => { }); }); +describe("createInvalidationBatcher with minimum spacing", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("flushes a quiet lane after the batch interval and spaces a busy lane", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300, { minSpacingMs: 5000 }); + batcher.schedule({ queryKey: ["issues", "c1"] }); + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(1); + // A continuous stream after that flush waits for the spacing window. + for (let elapsed = 0; elapsed < 4500; elapsed += 500) { + batcher.schedule({ queryKey: ["issues", "c1"] }); + vi.advanceTimersByTime(500); + } + expect(client.invalidateQueries).toHaveBeenCalledTimes(1); + vi.advanceTimersByTime(500); + expect(client.invalidateQueries).toHaveBeenCalledTimes(2); + }); + + it("returns to the batch interval once the spacing window has passed", () => { + const { client } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300, { minSpacingMs: 5000 }); + batcher.schedule({ queryKey: ["a"] }); + vi.advanceTimersByTime(300); + vi.advanceTimersByTime(10_000); + batcher.schedule({ queryKey: ["a"] }); + vi.advanceTimersByTime(300); + expect(client.invalidateQueries).toHaveBeenCalledTimes(2); + }); + + it("never drops an invalidation scheduled during the spacing window", async () => { + const { client, calls } = fakeClient(); + const batcher = createInvalidationBatcher(client, 300, { minSpacingMs: 5000 }); + batcher.schedule({ queryKey: ["a"] }); + vi.advanceTimersByTime(300); + const pending = batcher.schedule({ queryKey: ["b"] }); + await vi.advanceTimersByTimeAsync(5000); + await pending; + expect(calls).toEqual([{ queryKey: ["a"] }, { queryKey: ["b"] }]); + }); +}); + +describe("createLaneRoutingBatcher", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("routes ambient keys to the spaced lane and every other filter to the fast lane", () => { + const { client, calls } = fakeClient(); + const batcher = createLaneRoutingBatcher( + createInvalidationBatcher(client, 300), + createInvalidationBatcher(client, 300, { minSpacingMs: 5000 }), + (queryKey) => queryKey[0] === "issues" && queryKey[1] === "c1", + ); + batcher.schedule({ queryKey: ["issues", "c1"] }); + vi.advanceTimersByTime(300); + batcher.schedule({ queryKey: ["issues", "c1", "summaries"] }); + batcher.schedule({ queryKey: ["issues", "detail", "t1"] }); + batcher.schedule({}); + vi.advanceTimersByTime(300); + // The detail and key-less filters do not wait behind the ambient spacing. + expect(calls).toEqual([{ queryKey: ["issues", "c1"] }, { queryKey: ["issues", "detail", "t1"] }, {}]); + vi.advanceTimersByTime(5000); + expect(calls.at(-1)).toEqual({ queryKey: ["issues", "c1", "summaries"] }); + batcher.schedule({ queryKey: ["issues", "c1"] }); + batcher.dispose(); + vi.advanceTimersByTime(10_000); + expect(calls).toHaveLength(4); + }); +}); + describe("createCoalescingQueryClient", () => { beforeEach(() => vi.useFakeTimers()); afterEach(() => { vi.useRealTimers(); vi.unstubAllGlobals(); }); diff --git a/ui/src/lib/query-invalidation-batcher.ts b/ui/src/lib/query-invalidation-batcher.ts index af74a2ae51e..496893963dc 100644 --- a/ui/src/lib/query-invalidation-batcher.ts +++ b/ui/src/lib/query-invalidation-batcher.ts @@ -37,10 +37,24 @@ function createDeferred(): Deferred { return { promise, resolve }; } +/** + * Minimum time between two flushes of the ambient lane. Company-wide lists and + * projections (sidebar task summaries, the agent list, issue lists, badges) + * change on almost every run or task event in a busy company. After the lane + * has been quiet for this interval, an event refreshes them within one normal + * batch. An event inside the interval waits until the interval ends, so the + * lane flushes at most once per interval instead of on every batch. + */ +export const AMBIENT_INVALIDATION_SPACING_MS = 5_000; + export function createInvalidationBatcher( queryClient: Pick, intervalMs: number = DEFAULT_INVALIDATION_INTERVAL_MS, + options: { minSpacingMs?: number } = {}, ): InvalidationBatcher { + const minSpacingMs = options.minSpacingMs ?? 0; + // Start of the last flush that sent invalidations; spacing counts from it. + let lastFlushAt = Number.NEGATIVE_INFINITY; // Keyed by a stable serialization of the filters so repeated invalidations of // the same key (the common case) collapse to one entry. const pending = new Map(); @@ -62,6 +76,7 @@ export function createInvalidationBatcher( } const filtersList = removeCoveredFilters([...pending.values()]); pending.clear(); + lastFlushAt = Date.now(); try { await Promise.all(filtersList.map((filters) => queryClient.invalidateQueries( getPageVisibility().visible ? filters : { ...filters, refetchType: "none" }, @@ -77,7 +92,10 @@ export function createInvalidationBatcher( windowDeferred = createDeferred(); } if (timer === null) { - timer = setTimeout(() => void flush(), intervalMs); + // Leading edge: a quiet lane flushes after the normal batch interval. + // Trailing edge: a busy lane waits until its spacing has elapsed. + const delay = Math.max(intervalMs, lastFlushAt + minSpacingMs - Date.now()); + timer = setTimeout(() => void flush(), delay); } return windowDeferred.promise; }; @@ -226,3 +244,25 @@ export function createCoalescingQueryClient( }, }); } + +/** + * Route invalidations to one of two batchers by query key. `isAmbient` selects + * the slower lane (see `AMBIENT_INVALIDATION_SPACING_MS`). Every other filter, + * including one without a query key, uses the fast lane. + */ +export function createLaneRoutingBatcher( + fast: InvalidationBatcher, + ambient: InvalidationBatcher, + isAmbient: (queryKey: readonly unknown[]) => boolean, +): InvalidationBatcher { + return { + schedule: (filters) => (filters.queryKey && isAmbient(filters.queryKey) ? ambient : fast).schedule(filters), + flush: async () => { + await Promise.all([fast.flush(), ambient.flush()]); + }, + dispose: () => { + fast.dispose(); + ambient.dispose(); + }, + }; +} diff --git a/ui/src/lib/task-browsers-refresh.test.ts b/ui/src/lib/task-browsers-refresh.test.ts new file mode 100644 index 00000000000..ae5c931cc40 --- /dev/null +++ b/ui/src/lib/task-browsers-refresh.test.ts @@ -0,0 +1,99 @@ +// @vitest-environment jsdom +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { QueryClient, QueryObserver } from "@tanstack/react-query"; +import type { TaskBrowser } from "@paperclipai/shared"; +import { taskBrowsersRefetchInterval } from "../hooks/useTaskBrowsers"; +import { refreshTaskBrowsers, taskBrowsersQueryKey } from "./task-browsers-refresh"; + +const TASK = "12345678-1234-4234-8234-123456789abc"; +const started: TaskBrowser = { + id: "browser-1", sessionId: "session-1", issueId: TASK, status: "running", runStatus: "running", + progress: null, costCents: 0, idleDeadline: null, expiresAt: null, error: null, createdAt: "2026-11-09T10:00:00Z", +}; +let client: QueryClient; + +/** One mounted `useTaskBrowsers` with the live connection open. */ +function observe(list: () => Promise) { + const observer = new QueryObserver(client, { + queryKey: taskBrowsersQueryKey(TASK), queryFn: list, staleTime: 30_000, + refetchInterval: (query) => taskBrowsersRefetchInterval(true, query.state), + }); + return { observer, unsubscribe: observer.subscribe(() => {}) }; +} + +beforeEach(() => { + vi.useFakeTimers(); + client = new QueryClient({ defaultOptions: { queries: { retry: false } } }); +}); +afterEach(() => { + client.clear(); + vi.useRealTimers(); +}); + +it("reads again after a delayed empty read when the browser starts during that read", async () => { + let release!: (value: TaskBrowser[]) => void; + const list = vi.fn<() => Promise>() + .mockImplementationOnce(() => new Promise(resolve => { release = resolve; })) + .mockResolvedValue([started]); + const view = observe(list); + await vi.advanceTimersByTimeAsync(1); + refreshTaskBrowsers(client, TASK); + await vi.advanceTimersByTimeAsync(1); + // The refresh waits for the older read instead of joining it. + expect(list).toHaveBeenCalledTimes(1); + release([]); + await vi.advanceTimersByTimeAsync(1); + expect(list).toHaveBeenCalledTimes(2); + expect(view.observer.getCurrentResult().data).toEqual([started]); + view.unsubscribe(); +}); + +it("queues one follow-up for two views and repeated events during one read", async () => { + let release!: (value: TaskBrowser[]) => void; + const list = vi.fn<() => Promise>() + .mockImplementationOnce(() => new Promise(resolve => { release = resolve; })) + .mockResolvedValue([]); + const page = observe(list); + const panel = observe(list); + await vi.advanceTimersByTimeAsync(1); + refreshTaskBrowsers(client, TASK); + refreshTaskBrowsers(client, TASK); + release([]); + await vi.advanceTimersByTimeAsync(10_000); + expect(list).toHaveBeenCalledTimes(2); + page.unsubscribe(); + panel.unsubscribe(); +}); + +it("refreshes an idle list at once and stops polling when it stays empty", async () => { + const list = vi.fn(async () => [] as TaskBrowser[]); + const view = observe(list); + await vi.advanceTimersByTimeAsync(10_000); + expect(list).toHaveBeenCalledTimes(1); + refreshTaskBrowsers(client, TASK); + await vi.advanceTimersByTimeAsync(10_000); + expect(list).toHaveBeenCalledTimes(2); + view.unsubscribe(); +}); + +it("makes a returning view read a list cached while the task was closed", async () => { + const list = vi.fn<() => Promise>().mockResolvedValueOnce([]).mockResolvedValue([started]); + const first = observe(list); + await vi.advanceTimersByTimeAsync(1); + first.unsubscribe(); + refreshTaskBrowsers(client, TASK); + await vi.advanceTimersByTimeAsync(1); + // Nothing observes the list, so the refresh only marks it stale. + expect(list).toHaveBeenCalledTimes(1); + const returning = observe(list); + await vi.advanceTimersByTimeAsync(1); + expect(list).toHaveBeenCalledTimes(2); + expect(returning.observer.getCurrentResult().data).toEqual([started]); + returning.unsubscribe(); +}); + +it("does nothing for a task without a cached browser list", () => { + const invalidate = vi.spyOn(client, "invalidateQueries"); + refreshTaskBrowsers(client, TASK); + expect(invalidate).not.toHaveBeenCalled(); +}); diff --git a/ui/src/lib/task-browsers-refresh.ts b/ui/src/lib/task-browsers-refresh.ts new file mode 100644 index 00000000000..15caff7a496 --- /dev/null +++ b/ui/src/lib/task-browsers-refresh.ts @@ -0,0 +1,55 @@ +import type { QueryClient } from "@tanstack/react-query"; + +export function taskBrowsersQueryKey(issueId: string | undefined) { + return ["task-browsers", issueId] as const; +} + +// One queued follow-up per cached browser list, shared by every caller. +const queuedFollowUps = new WeakMap>(); + +/** + * Refresh a task's browser list after browser activity on that task. + * + * The list stops polling while the live connection is open and no browser can + * change state, so this activity is its refresh signal. A read already in + * flight may predate the activity, and a refresh that joined it could keep an + * older empty list. In that case one follow-up read runs after it settles. + * Invalidation also marks a list that no view observes as stale, so a view + * that opens the task again reads it. + */ +export function refreshTaskBrowsers(queryClient: QueryClient, issueId: string): void { + refreshTaskBrowserList(queryClient, taskBrowsersQueryKey(issueId)); +} + +/** + * Queue a follow-up for every browser list with a read in flight when the live + * connection opens. Events missed while it was closed are not replayed, and + * the connection-open reconcile can join a read that started before the + * missed activity. Lists without a read in flight are left to that reconcile. + */ +export function refreshPendingTaskBrowserReads(queryClient: QueryClient): void { + for (const query of queryClient.getQueryCache().findAll({ queryKey: ["task-browsers"] })) { + if (query.state.fetchStatus === "fetching") refreshTaskBrowserList(queryClient, query.queryKey); + } +} + +function refreshTaskBrowserList(queryClient: QueryClient, queryKey: readonly unknown[]): void { + const cache = queryClient.getQueryCache(); + const query = cache.find({ queryKey, exact: true }); + if (!query) return; + const invalidate = () => queryClient.invalidateQueries({ queryKey, exact: true }); + const pending = query.state.fetchStatus === "fetching" ? query.promise : undefined; + if (!pending) { + void invalidate(); + return; + } + const queued = queuedFollowUps.get(cache) ?? new Set(); + queuedFollowUps.set(cache, queued); + const id = JSON.stringify(queryKey); + if (queued.has(id)) return; + queued.add(id); + void pending.catch(() => undefined).then(() => { + queued.delete(id); + return invalidate(); + }); +} diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index d710a3229ae..11af0e50932 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -4078,7 +4078,7 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS if (isMobile) setMobilePropsOpen(true); } }, [pluginPanels, taskPanelAccountScope, issue?.companyId, issue?.id, openTaskSidePanel, isMobile]); - const browserQuery = useTaskBrowsers(issue?.id); + const browserQuery = useTaskBrowsers(issue?.id, issue?.companyId); const { openBrowserId, openBrowser: setOpenBrowserId, acknowledgeBrowserOpened: handleBrowserOpened } = useBrowserArrivals(currentUserId ?? "anonymous", issue?.id, browserQuery.data); useEffect(() => {