From d82e5efeb983c0c497cdc9e884875c5919aedb21 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 11:01:53 -0600 Subject: [PATCH] perf(ui): cut repeated reads while a task is open (#20563) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The board UI keeps an open task current with a shared company websocket and with polling fallbacks > - In a busy company, one open task sent 35 to 60 API requests in 17 seconds, and each request waits on the database > - Most of these requests were repeated reads: task summaries 22 times, the task run list 11 times, task browsers 6 times, and the 60 KB agent list 6 times > - The cause is not the open task. Other agents' run and task events invalidate company-wide lists on every 300 ms batch, and the task view polls runs and browsers on fixed intervals even while the websocket is open > - This pull request spaces company-wide refreshes, and it lets the websocket replace run and browser polling while it is connected > - The benefit is about 79% fewer reads from an open task in a busy company, with the same polling fallback when the connection is down ## Linked Issues or Issue Description Refs #17608 Refs #9627 **What happened?** An open task in a busy company sent the same reads again and again. In one 17-second visit, the UI requested `issues/summaries` 22 times, the task live runs 11 times, the task browsers 6 times, and the company agent list 6 times. The company live runs and the task's child lists (`parentId`, `createdFromIssueId`, `descendantOf`) were requested 2 to 3 times each. **Expected behavior** A busy company must not make an open task refetch company-wide lists on every event batch. While the live connection is open, the UI must use events, not fixed fast intervals, for task runs and browsers. The UI must keep its polling fallback when the connection is closed. **Steps to reproduce** 1. Open a task in a company where other agents run work. 2. Keep the task open for 17 seconds. 3. Count the API requests in the browser network panel. **Paperclip version or commit** Based on master at bd30ef8510. ## What Changed - `ui/src/lib/query-invalidation-batcher.ts`: Add a minimum spacing option to the invalidation batcher. A quiet lane flushes after the normal batch interval. A busy lane flushes at most once per spacing window. Add `createLaneRoutingBatcher`, which sends each invalidation to one of two batchers by query key. - `ui/src/context/LiveUpdatesProvider.tsx`: Route company-wide keys to a 5-second ambient lane. These keys are task lists and summaries, the agent list, heartbeats, company live runs, sidebar badges, the dashboard, costs and the company activity feed. All other keys, such as task detail, comments, task runs and browsers, keep the 300 ms lane. - `ui/src/lib/live-runs-cache.ts`: While the company connection is open, a successful active run list or active run polls every 15 seconds instead of every 1 or 3 seconds. Run lifecycle events already refresh these queries, and progress events patch them. - `ui/src/hooks/useTaskBrowsers.ts`: While the company connection is open, stop browser polling when no browser can change state. Keep the 3-second poll for a starting, running, idle or stopping browser, and when the connection is closed. The two callers pass the task's company. Socket open and reconnect already reconcile this active query. - `ui/src/lib/task-browsers-refresh.ts` and `LiveUpdatesProvider.tsx`: On `browser.*` activity for a task, refresh that task's browser list. If a read is in flight, one shared follow-up read runs after it settles, so an older empty result cannot stop polling. Two views of the same task share that follow-up. A list that no view observes becomes stale, so a returning view reads it. When the connection opens or resyncs, each browser list with a read in flight also gets one follow-up read, because missed activity is not replayed. - `doc/DATABASE.md`: Document the ambient lane and the new polling contract. - Tests: Add a fan-out test that replays a busy-company event stream for 17 seconds against an open task, plus unit tests for spacing, lane routing, key classification, the run interval and the browser interval. Update the recovery test for the connected active-run interval. ## Verification Request count over a fixed 17-second open window (`ui/src/context/LiveUpdatesProvider.fanout.test.ts`). The test mounts the reads that an open, running task keeps active. It replays a busy company: one run status event every second and one task activity event every 0.7 seconds, none for the open task. Fake timers make each run deterministic, so the spread is zero. The baseline used the same test with master's batcher, run interval and browser interval. | Read | Before | After | | --- | ---: | ---: | | Task summaries (sidebar) | 24 | 5 | | Company agent list | 17 | 5 | | Company live runs | 17 | 5 | | Child list: parent | 24 | 5 | | Child list: created from | 24 | 5 | | Child list: descendants | 24 | 5 | | Task live runs | 18 | 2 | | Task browsers | 6 | 1 | | **Total** | **154** | **33** | The baseline numbers match the staging capture well (summaries 22, browsers 6). - `pnpm exec vitest run src/context/LiveUpdatesProvider*.test.* src/lib/live-runs-cache.test.ts src/lib/query-invalidation-batcher.test.ts src/hooks/useTaskBrowsers.test.ts src/lib/task-browsers-refresh.test.ts src/hooks/useTaskPluginPanels.test.tsx src/components/task-side-panel/TaskSidePanel.test.tsx src/pages/IssueDetail.test.tsx` (in `ui/`): all pass. - `pnpm exec tsc --noEmit -p ui`: no errors in `ui/src`. - `pnpm check:token-gates`: clean. - Tests also prove these cases: - With the connection closed, task runs still poll every second and browsers every 3 seconds. - With a quiet company, one event refreshes the summaries and child lists after one normal batch. - `browser.started` on the open task refreshes its browsers. The same event on another task does not. - A browser that starts during a slow empty read appears after that read settles (`ui/src/lib/task-browsers-refresh.test.ts`). This test fails when the refresh joins the older read. - Two views and repeated events during one read cause one follow-up read. This test fails without the shared queue. - A list cached while the task was closed is read again when a view returns. - A connection that opens during an older empty browser read causes one more read, so a browser that started while the socket was closed appears. This test fails without the follow-up. - Task detail, comments, task runs, browsers, attention and another company's keys stay on the fast lane. - An unrecognized active result keeps the fallback interval. This is a local request-count proof. It does not include a timed browser journey on a deployed instance. ## Risks - Company-wide lists refresh at most once every 5 seconds from live events, so they can be up to 5 seconds late. After 5 quiet seconds, an event refreshes them in about 300 ms. Own mutations invalidate their queries directly, so they do not wait. - An active task run refreshes from events. Fields without an event, such as output silence, can be up to 15 seconds late while the connection is open. Reconnect still reconciles missed events. - A browser that appears without activity is not shown until the next activity or reconnect. Every browser lifecycle step logs activity on the task, so this case is not expected. - No server, schema or API change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/DATABASE.md | 19 +- .../task-side-panel/TaskSidePanel.tsx | 2 +- .../LiveUpdatesProvider.fanout.test.ts | 206 ++++++++++++++++++ .../LiveUpdatesProvider.recovery.test.tsx | 83 ++++++- ui/src/context/LiveUpdatesProvider.tsx | 56 ++++- ui/src/hooks/useTaskBrowsers.test.ts | 17 +- ui/src/hooks/useTaskBrowsers.ts | 35 ++- ui/src/lib/live-runs-cache.test.ts | 15 +- ui/src/lib/live-runs-cache.ts | 23 +- ui/src/lib/query-invalidation-batcher.test.ts | 72 ++++++ ui/src/lib/query-invalidation-batcher.ts | 42 +++- ui/src/lib/task-browsers-refresh.test.ts | 99 +++++++++ ui/src/lib/task-browsers-refresh.ts | 55 +++++ ui/src/pages/IssueDetail.tsx | 2 +- 14 files changed, 706 insertions(+), 20 deletions(-) create mode 100644 ui/src/context/LiveUpdatesProvider.fanout.test.ts create mode 100644 ui/src/lib/task-browsers-refresh.test.ts create mode 100644 ui/src/lib/task-browsers-refresh.ts 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(() => {