From e33526556879e0720e95cd112a5faa3bffac8ce2 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 25 Oct 2026 07:54:47 -0700 Subject: [PATCH] fix(ui): stop idle task run polling with live updates (#17608) Stop successful empty assigned-task run queries from polling only while the matching company and session connection is open. Preserve active, pending, failed, and disconnected cadence. Share one lifecycle refresh owner between the task and chat views, discover reviewer events without task identity, and refresh after older in-flight reads settle. All 175 focused tests, 9,079 UI tests, build, typecheck, exact-head Greptile 5/5, and required CI pass. Regression mutations catch reviewer discovery, duplicate reads, readiness, failure state, missing refreshes, and scope cleanup. Earlier E2E cleanup deadlock and server cleanup timeout remain documented; failed-only reruns passed on their same commits. Co-Authored-By: Paperclip --- doc/DATABASE.md | 15 + .../LiveUpdatesProvider.recovery.test.tsx | 359 +++++++++++++++++- ui/src/context/LiveUpdatesProvider.tsx | 86 ++++- ui/src/lib/live-runs-cache.test.ts | 18 +- ui/src/lib/live-runs-cache.ts | 10 + ui/src/pages/IssueDetail.tsx | 14 +- 6 files changed, 481 insertions(+), 21 deletions(-) diff --git a/doc/DATABASE.md b/doc/DATABASE.md index 780247fc5c8..8056e2468c3 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -145,6 +145,21 @@ patch does not refetch task details, activity, or run history. Unknown runs 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 +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 +focus. Events that omit task identity conservatively refresh these queries, +including reviewer runs from another agent. One parent view owns both query +observers and reuses refreshes already covered by the focused-route handler. A +matching event that overlaps an older request queues a fresh read after that +request settles. Connection readiness also reconciles the displayed run +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. + Connection-assignment checks select profile bindings for the current company or agent and read rules that can match the selected connection's catalog rows. All five selector kinds, null application matching, exclusions, and profile diff --git a/ui/src/context/LiveUpdatesProvider.recovery.test.tsx b/ui/src/context/LiveUpdatesProvider.recovery.test.tsx index b63381e9b44..591e735fa49 100644 --- a/ui/src/context/LiveUpdatesProvider.recovery.test.tsx +++ b/ui/src/context/LiveUpdatesProvider.recovery.test.tsx @@ -5,18 +5,19 @@ import { QueryClient, QueryClientProvider, useQuery } from "@tanstack/react-quer import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { queryKeys } from "../lib/queryKeys"; -const { pushToast } = vi.hoisted(() => ({ pushToast: vi.fn() })); +const { pushToast, companySelection, routeSelection } = vi.hoisted(() => ({ pushToast: vi.fn(), companySelection: { id: "company-1" }, routeSelection: { pathname: "/tasks" } })); const { sessionRead, healthRead } = vi.hoisted(() => ({ sessionRead: vi.fn(), healthRead: vi.fn() })); vi.mock("./CompanyContext", () => ({ - useCompany: () => ({ selectedCompanyId: "company-1", selectedCompany: { id: "company-1" } }), + useCompany: () => ({ selectedCompanyId: companySelection.id, selectedCompany: { id: companySelection.id } }), })); vi.mock("./ToastContext", () => ({ useToastActions: () => ({ pushToast }) })); -vi.mock("../lib/router", () => ({ useLocation: () => ({ pathname: "/tasks" }) })); +vi.mock("../lib/router", () => ({ useLocation: () => routeSelection })); vi.mock("../api/auth", () => ({ authApi: { getSession: sessionRead }, })); vi.mock("../api/health", () => ({ healthApi: { get: healthRead } })); -import { LiveUpdatesProvider } from "./LiveUpdatesProvider"; +import { LiveUpdatesProvider, useHasCompanyLiveConnection, useTaskRunLiveUpdates } from "./LiveUpdatesProvider"; +import { liveRunRefetchInterval } from "../lib/live-runs-cache"; (globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; @@ -25,8 +26,8 @@ class Socket { readyState = 1; onopen: (() => void) | null = null; onclose: (() => void) | null = null; - onmessage = null; - onerror = null; + onmessage: ((message: { data: string }) => void) | null = null; + onerror: (() => void) | null = null; constructor() { Socket.instances.push(this); } close() { this.readyState = 3; } } @@ -36,9 +37,13 @@ describe("LiveUpdatesProvider connection recovery", () => { let container: HTMLDivElement; let client: QueryClient; const read = vi.fn(async () => "current task data"); + const pollRead = vi.fn<() => Promise>(); beforeEach(() => { vi.useFakeTimers(); + companySelection.id = "company-1"; + routeSelection.pathname = "/tasks"; + pollRead.mockReset().mockResolvedValue([]); vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible"); read.mockReset(); read.mockResolvedValue("current task data"); @@ -71,6 +76,348 @@ describe("LiveUpdatesProvider connection recovery", () => { await act(async () => vi.advanceTimersByTimeAsync(1)); } + function IdleRuns({ companyId }: { companyId: string }) { + const connected = useHasCompanyLiveConnection(companyId); + useQuery({ + queryKey: [companyId, "idle-run-fixture"], + queryFn: pollRead, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, 1000), + }); + return {connected ? "connected" : "fallback"}; + } + + async function renderPolling(companyId = companySelection.id, withProvider = true) { + const child = ; + await act(async () => root.render( + + {withProvider ? {child} : child} + , + )); + await act(async () => vi.advanceTimersByTimeAsync(1)); + } + + async function advancePolls(count: number) { + for (let i = 0; i < count; i += 1) { + await act(async () => vi.advanceTimersByTimeAsync(1001)); + } + } + + it.each([null, []])("keeps setup polling, then stops known-empty polling after connection for %j", async (data) => { + vi.stubGlobal("WebSocket", Socket); + pollRead.mockResolvedValue(data); + await renderPolling(); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + await advancePolls(3); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(3); + await act(async () => Socket.instances[0].onopen?.()); + await act(async () => vi.advanceTimersByTimeAsync(1)); + expect(container.textContent).toBe("connected"); + const readsAfterOpen = pollRead.mock.calls.length; + await advancePolls(10); + expect(pollRead).toHaveBeenCalledTimes(readsAfterOpen); + }); + + it.each([[{ id: "run-1" }], { id: "run-1" }])("keeps active work polling while connected for %j", async (data) => { + vi.stubGlobal("WebSocket", Socket); + pollRead.mockResolvedValue(data); + await renderPolling(); + await act(async () => Socket.instances[0].onopen?.()); + pollRead.mockClear(); + await advancePolls(3); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(3); + }); + + it("keeps polling a failed query with previously empty data until it recovers", async () => { + vi.stubGlobal("WebSocket", Socket); + await renderPolling(); + await act(async () => Socket.instances[0].onopen?.()); + pollRead.mockRejectedValue(new Error("Read unavailable")); + await act(async () => { await client.invalidateQueries({ queryKey: ["company-1", "idle-run-fixture"] }); }); + expect(client.getQueryState(["company-1", "idle-run-fixture"])?.status).toBe("error"); + pollRead.mockClear(); + await advancePolls(3); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(3); + pollRead.mockResolvedValue([]); + await advancePolls(1); + expect(client.getQueryState(["company-1", "idle-run-fixture"])?.status).toBe("success"); + const afterRecovery = pollRead.mock.calls.length; + await advancePolls(5); + expect(pollRead).toHaveBeenCalledTimes(afterRecovery); + }); + + it("retains an initial pending read until it resolves after the connection opens", async () => { + vi.stubGlobal("WebSocket", Socket); + let resolveRead!: (value: unknown) => void; + pollRead.mockImplementation(() => new Promise((resolve) => { resolveRead = resolve; })); + await renderPolling(); + await act(async () => Socket.instances[0].onopen?.()); + expect(container.textContent).toBe("connected"); + expect(client.getQueryState(["company-1", "idle-run-fixture"])?.status).toBe("pending"); + expect(pollRead).toHaveBeenCalledTimes(1); + await act(async () => resolveRead([])); + await act(async () => vi.advanceTimersByTimeAsync(1)); + expect(client.getQueryData(["company-1", "idle-run-fixture"])).toEqual([]); + await advancePolls(5); + expect(pollRead).toHaveBeenCalledTimes(1); + }); + + it("resumes local polling on error before close, and stops it after recovery", async () => { + vi.stubGlobal("WebSocket", Socket); + await renderPolling(); + const first = Socket.instances[0]; + await act(async () => first.onopen?.()); + await act(async () => first.onerror?.()); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + await advancePolls(3); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(3); + await act(async () => first.onclose?.()); + await advancePolls(2); + expect(Socket.instances.length).toBeGreaterThan(1); + await act(async () => Socket.instances.at(-1)?.onopen?.()); + expect(container.textContent).toBe("connected"); + const afterRecovery = pollRead.mock.calls.length; + await advancePolls(10); + expect(pollRead).toHaveBeenCalledTimes(afterRecovery); + }); + + it.each(["company", "session"])("requires a new connection after a %s change and ignores stale callbacks", async (scope) => { + vi.stubGlobal("WebSocket", Socket); + await renderPolling(); + const first = Socket.instances[0]; + await act(async () => first.onopen?.()); + const oldOpen = first.onopen, oldClose = first.onclose, oldError = first.onerror; + if (scope === "company") { + companySelection.id = "company-2"; + await renderPolling(); + } else { + const nextSession = { user: { id: "viewer" }, session: { id: "replacement-session", userId: "viewer" } }; + sessionRead.mockResolvedValue(nextSession); + await act(async () => client.setQueryData(queryKeys.auth.session, nextSession)); + await act(async () => vi.advanceTimersByTimeAsync(1)); + } + expect(container.textContent).toBe("fallback"); + await act(async () => oldOpen?.()); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + await advancePolls(2); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(2); + await act(async () => Socket.instances.at(-1)?.onopen?.()); + expect(container.textContent).toBe("connected"); + await act(async () => { oldClose?.(); oldError?.(); }); + expect(container.textContent).toBe("connected"); + const afterOpen = pollRead.mock.calls.length; + await advancePolls(5); + expect(pollRead).toHaveBeenCalledTimes(afterOpen); + }); + + it("discovers matching new runs after empty polling stops on an unfocused non-task route", async () => { + vi.stubGlobal("WebSocket", Socket); + vi.spyOn(document, "hasFocus").mockReturnValue(false); + const queryRef = "DEMO-1"; + const runRead = vi.fn<() => Promise>().mockResolvedValue([]); + const activeRead = vi.fn<() => Promise>().mockResolvedValue(null); + function TaskRuns() { + const connected = useTaskRunLiveUpdates({ companyId: "company-1", issueId: "issue-1", issueQueryRef: queryRef, assigneeAgentId: "agent-1" }); + const runs = useQuery({ queryKey: queryKeys.issues.liveRuns(queryRef), queryFn: runRead, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, 1000) }); + useQuery({ queryKey: queryKeys.issues.activeRun(queryRef), queryFn: activeRead, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, 1000) }); + return {JSON.stringify(runs.data)}; + } + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTimeAsync(1)); + const socket = Socket.instances[0]; + await act(async () => socket.onopen?.()); + runRead.mockClear(); activeRead.mockClear(); + await advancePolls(5); + expect(runRead).not.toHaveBeenCalled(); + expect(activeRead).not.toHaveBeenCalled(); + const deliver = async (companyId: string, payload: Record) => { + await act(async () => socket.onmessage?.({ data: JSON.stringify({ id: 1, companyId, + type: "heartbeat.run.queued", createdAt: "2020-01-01T00:00:00Z", payload }) })); + await act(async () => vi.advanceTimersByTimeAsync(1)); + }; + await deliver("company-2", { issueId: "issue-1", agentId: "agent-1", runId: "foreign" }); + await deliver("company-1", { issueId: "other-issue", agentId: "agent-1", runId: "other" }); + expect(runRead).not.toHaveBeenCalled(); + expect(activeRead).not.toHaveBeenCalled(); + runRead.mockResolvedValue([{ id: "new-run", status: "running" }]); + activeRead.mockResolvedValue({ id: "new-run", status: "running" }); + await deliver("company-1", { issueId: "issue-1", agentId: "new-agent", runId: "new-run" }); + expect(runRead).toHaveBeenCalledTimes(1); + expect(activeRead).toHaveBeenCalledTimes(1); + expect(container.textContent).toContain("new-run"); + await deliver("company-1", { agentId: "reviewer-agent", runId: "reviewer-run-without-task-id" }); + expect(runRead).toHaveBeenCalledTimes(2); + expect(activeRead).toHaveBeenCalledTimes(2); + }); + + it.each(["initial", "background", "connection setup"])("refetches after an older empty %s read settles when run events arrive during it", async (phase) => { + vi.stubGlobal("WebSocket", Socket); + vi.spyOn(document, "hasFocus").mockReturnValue(false); + const queryKey = queryKeys.issues.liveRuns("DEMO-1"); + const runRead = vi.fn<() => Promise>().mockResolvedValue([]); + let resolveOlder!: (value: unknown) => void; + const older = new Promise((resolve) => { resolveOlder = resolve; }); + if (phase !== "background") runRead.mockImplementationOnce(() => older); + function TaskRuns() { + const connected = useTaskRunLiveUpdates({ companyId: "company-1", issueId: "issue-1", issueQueryRef: "DEMO-1", assigneeAgentId: "agent-1" }); + const { data } = useQuery({ queryKey, queryFn: runRead, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, 1000) }); + return {JSON.stringify(data)}; + } + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTimeAsync(1)); + const socket = Socket.instances[0]; + await act(async () => socket.onopen?.()); + if (phase === "background") { + runRead.mockClear(); + runRead.mockImplementationOnce(() => older); + await act(async () => { void client.invalidateQueries({ queryKey }); }); + } + runRead.mockResolvedValue([{ id: "new-run", status: "running" }]); + expect(runRead).toHaveBeenCalledTimes(1); + await act(async () => { + for (let id = 1; phase !== "connection setup" && id <= 3; id += 1) socket.onmessage?.({ data: JSON.stringify({ id, + companyId: "company-1", type: "heartbeat.run.queued", createdAt: "2020-01-01T00:00:00Z", + payload: { issueId: "issue-1", agentId: "agent-1", runId: "new-run" } }) }); + }); + expect(runRead).toHaveBeenCalledTimes(1); + await act(async () => resolveOlder([])); + await act(async () => vi.advanceTimersByTimeAsync(1)); + expect(runRead).toHaveBeenCalledTimes(2); + expect(client.getQueryData(queryKey)).toEqual([{ id: "new-run", status: "running" }]); + expect(container.textContent).toContain("new-run"); + }); + + it.each(["unmount", "disconnect", "company", "session", "task"])("discards the old follow-up refresh after %s during an older read", async (change) => { + vi.stubGlobal("WebSocket", Socket); + let resolveOlder!: (value: unknown) => void; + const older = new Promise((resolve) => { resolveOlder = resolve; }); + const runRead = vi.fn<() => Promise>().mockResolvedValue([]).mockImplementationOnce(() => older); + function TaskRuns({ task = "1" }: { task?: string }) { + const queryRef = `DEMO-${task}`; + const connected = useTaskRunLiveUpdates({ companyId: "company-1", issueId: `issue-${task}`, issueQueryRef: queryRef, assigneeAgentId: "agent-1" }); + useQuery({ queryKey: queryKeys.issues.liveRuns(queryRef), queryFn: runRead, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, 1000) }); + return null; + } + const render = async (task = "1") => { + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTimeAsync(1)); + }; + await render(); + const socket = Socket.instances[0]; + await act(async () => socket.onopen?.()); + await act(async () => socket.onmessage?.({ data: JSON.stringify({ id: 1, companyId: "company-1", + type: "heartbeat.run.queued", createdAt: "2020-01-01T00:00:00Z", + payload: { issueId: "issue-1", agentId: "agent-1", runId: "new-run" } }) })); + if (change === "unmount") await act(async () => root.unmount()); + if (change === "disconnect") await act(async () => socket.onerror?.()); + if (change === "company") { companySelection.id = "company-2"; await render(); } + if (change === "task") await render("2"); + if (change === "session") { + const nextSession = { user: { id: "viewer" }, session: { id: "replacement-session", userId: "viewer" } }; + sessionRead.mockResolvedValue(nextSession); + await act(async () => client.setQueryData(queryKeys.auth.session, nextSession)); + await act(async () => vi.advanceTimersByTimeAsync(1)); + } + const readsAfterChange = runRead.mock.calls.length; + const invalidate = vi.spyOn(client, "invalidateQueries"); + await act(async () => resolveOlder([])); + await act(async () => vi.advanceTimersByTimeAsync(1)); + expect(runRead).toHaveBeenCalledTimes(readsAfterChange); + expect(invalidate).not.toHaveBeenCalled(); + }); + + it.each(["focused", "unfocused"])("refreshes once for a start event with two run observers on a %s task view", async focus => { + vi.stubGlobal("WebSocket", Socket); + vi.spyOn(document, "hasFocus").mockReturnValue(focus === "focused"); + routeSelection.pathname = "/DEMO/issues/DEMO-1"; + const queryRef = "DEMO-1"; + client.setQueryData(queryKeys.issues.detail(queryRef), { id: "issue-1", companyId: "company-1", assigneeAgentId: "agent-1" }); + const runRead = vi.fn<() => Promise>().mockResolvedValue([]); + const activeRead = vi.fn<() => Promise>().mockResolvedValue(null); + function Reads({ connected, interval }: { connected: boolean; interval: number }) { + useQuery({ queryKey: queryKeys.issues.liveRuns(queryRef), queryFn: runRead, + refetchInterval: query => liveRunRefetchInterval(connected, query.state, interval) }); + useQuery({ queryKey: queryKeys.issues.activeRun(queryRef), queryFn: activeRead, + refetchInterval: query => liveRunRefetchInterval(connected, query.state, interval) }); + return null; + } + function TaskRuns() { + const connected = useTaskRunLiveUpdates({ companyId: "company-1", issueId: "issue-1", issueQueryRef: queryRef, assigneeAgentId: "agent-1" }); + return <>; + } + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTimeAsync(1)); + const socket = Socket.instances[0]; + await act(async () => socket.onopen?.()); + await act(async () => vi.advanceTimersByTimeAsync(301)); + runRead.mockClear(); activeRead.mockClear(); + await act(async () => socket.onmessage?.({ data: JSON.stringify({ id: 1, companyId: "company-1", + type: "heartbeat.run.queued", createdAt: "2020-01-01T00:00:00Z", + payload: { agentId: "agent-1", runId: "new-run" } }) })); + await act(async () => vi.advanceTimersByTimeAsync(301)); + expect(runRead).toHaveBeenCalledTimes(1); + expect(activeRead).toHaveBeenCalledTimes(1); + }); + + it("keeps polling unassigned tasks even with a healthy shared connection", async () => { + vi.stubGlobal("WebSocket", Socket); + function UnassignedTask() { + const connected = useTaskRunLiveUpdates({ companyId: "company-1", issueId: "issue-1", issueQueryRef: "issue-1", assigneeAgentId: null }); + useQuery({ queryKey: queryKeys.issues.liveRuns("issue-1"), queryFn: pollRead, + refetchInterval: (query) => liveRunRefetchInterval(connected, query.state, 1000) }); + return {connected ? "connected" : "fallback"}; + } + await act(async () => root.render()); + await act(async () => vi.advanceTimersByTimeAsync(1)); + await act(async () => Socket.instances[0].onopen?.()); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + await advancePolls(2); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(2); + }); + + it("retains polling for another company and outside the provider", async () => { + vi.stubGlobal("WebSocket", Socket); + await renderPolling("company-2"); + await act(async () => Socket.instances[0].onopen?.()); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + await advancePolls(2); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(2); + await renderPolling("company-2", false); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + await advancePolls(2); + expect(pollRead.mock.calls.length).toBeGreaterThanOrEqual(2); + }); + + it("drops readiness while hidden and does not poll or retry after unmount", async () => { + vi.stubGlobal("WebSocket", Socket); + await renderPolling(); + await act(async () => Socket.instances[0].onopen?.()); + const visibility = vi.spyOn(document, "visibilityState", "get"); + await act(async () => { + visibility.mockReturnValue("hidden"); + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(container.textContent).toBe("fallback"); + pollRead.mockClear(); + const count = Socket.instances.length; + await advancePolls(5); + expect(pollRead).not.toHaveBeenCalled(); + expect(Socket.instances).toHaveLength(count); + await act(async () => root.unmount()); + await advancePolls(5); + expect(pollRead).not.toHaveBeenCalled(); + expect(Socket.instances).toHaveLength(count); + }); + it("reconciles updates between the initial query and the first socket connection", async () => { vi.stubGlobal("WebSocket", Socket); await render(); diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index 5f634a8cdde..d24879ba378 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -7,6 +7,7 @@ import { useEffect, useMemo, useRef, + useState, type ReactNode, } from "react"; import { @@ -87,6 +88,13 @@ interface LiveEventSubscription { const LiveEventSubscriptionContext = createContext(null); +const CompanyLiveConnectionContext = createContext(null); + +export function useHasCompanyLiveConnection(companyId: string | null | undefined): boolean { + const connectedCompanyId = useContext(CompanyLiveConnectionContext); + return Boolean(companyId && connectedCompanyId === companyId); +} + function dispatchLiveEventToSubscribers( subscribers: Set, expectedCompanyId: string, @@ -126,6 +134,51 @@ export function useCompanyLiveEvent(handler: CompanyLiveEventHandler): void { }, [subscription]); } +export function useTaskRunLiveUpdates(scope: { + companyId: string | null | undefined; + issueId: string | null | undefined; + issueQueryRef: string; + assigneeAgentId: string | null | undefined; +}): boolean { + const connected = useHasCompanyLiveConnection(scope.companyId); + const queryClient = useQueryClient(); + const location = useLocation(); + const refreshOwner = useRef<{ active: boolean } | null>(null); + const refreshRuns = useCallback((coveredByVisibleRoute = false) => { + const owner = refreshOwner.current; + if (!owner?.active) return; + for (const queryKey of [queryKeys.issues.liveRuns(scope.issueQueryRef), queryKeys.issues.activeRun(scope.issueQueryRef)]) { + const query = queryClient.getQueryCache().find({ queryKey }); + if (query?.state.fetchStatus === "fetching") { + void (query.promise ?? queryClient.invalidateQueries({ queryKey }, { cancelRefetch: false })) + .catch(() => undefined).then(() => { + if (owner.active) return queryClient.invalidateQueries({ queryKey }, { cancelRefetch: false }); + }); + } else if (!coveredByVisibleRoute) { + void queryClient.invalidateQueries({ queryKey }, { cancelRefetch: false }); + } + } + }, [queryClient, scope.issueQueryRef]); + useEffect(() => { + const owner = { active: true }; + refreshOwner.current = owner; + if (connected && scope.issueId && scope.assigneeAgentId) refreshRuns(); + return () => { owner.active = false; }; + }, [connected, refreshRuns, scope.companyId, scope.issueId, scope.assigneeAgentId]); + useCompanyLiveEvent((event) => { + if (event.companyId !== scope.companyId || !scope.issueId || + (event.type !== "heartbeat.run.queued" && event.type !== "heartbeat.run.status")) return; + const issueId = readString(event.payload.issueId); + const matches = issueId + ? issueId === scope.issueId || issueId === scope.issueQueryRef + : true; + if (!matches) return; + const visible = resolveVisibleIssueRouteContext(queryClient, location.pathname); + refreshRuns(Boolean(visible?.issueRefs.has(scope.issueQueryRef) && runEventMatchesVisibleIssue(visible, event.payload))); + }); + return connected && Boolean(scope.issueId && scope.assigneeAgentId); +} + function readString(value: unknown): string | null { return typeof value === "string" && value.length > 0 ? value : null; } @@ -476,6 +529,13 @@ function shouldSuppressRunStatusToastForVisibleIssue( ); } +function runEventMatchesVisibleIssue(context: VisibleIssueRouteContext, payload: Record): boolean { + const runId = readString(payload.runId); + const agentId = readString(payload.agentId); + return (runId !== null && context.runIds.has(runId)) || + Boolean(agentId && context.assigneeAgentId && agentId === context.assigneeAgentId); +} + function invalidateVisibleIssueRunQueries( queryClient: QueryClient, pathname: string, @@ -490,13 +550,7 @@ function invalidateVisibleIssueRunQueries( if (!context) return false; const runId = readString(payload.runId); - const agentId = readString(payload.agentId); - const matchesVisibleIssue = - (runId !== null && context.runIds.has(runId)) || - (!!agentId && - !!context.assigneeAgentId && - agentId === context.assigneeAgentId); - if (!matchesVisibleIssue) return false; + if (!runEventMatchesVisibleIssue(context, payload)) return false; const status = readString(payload.status); if (runId && status && TERMINAL_RUN_STATUSES.has(status)) { @@ -1988,7 +2042,9 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { const currentUserId = session?.user?.id ?? session?.session?.userId ?? (health?.deploymentMode === "local_trusted" ? "local-board" : null); const socketAuthKey = session?.session?.id ?? currentUserId ?? "signed_out"; + const [connectedScopeKey, setConnectedScopeKey] = useState(null); const liveCompanyId = resolveLiveCompanyId(selectedCompanyId, selectedCompany?.id ?? null); + const socketScopeKey = JSON.stringify([liveCompanyId, socketAuthKey]); const canConnectSocket = canUseLiveSession(sessionStatus, session != null, health?.deploymentMode) && liveCompanyId !== null; const currentActorRef = useRef<{ userId: string | null; agentId: string | null }>({ userId: currentUserId, @@ -2104,6 +2160,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { closeSocketQuietly(nextSocket, "stale_connection"); return; } + setConnectedScopeKey(socketScopeKey); stopPolling(); if (reconnectAttempt > 0) { gateRef.current.suppressUntil = Date.now() + RECONNECT_SUPPRESS_MS; @@ -2155,6 +2212,8 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { }; nextSocket.onerror = () => { + if (closed || socket !== nextSocket) return; + setConnectedScopeKey((current) => current === socketScopeKey ? null : current); // Wait for onclose to drive the reconnect. Self-closing here is what // produces the "closed before connection established" browser noise. }; @@ -2163,6 +2222,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { if (socket !== nextSocket) return; socket = null; if (closed) return; + setConnectedScopeKey((current) => current === socketScopeKey ? null : current); startPolling(); scheduleReconnect(); }; @@ -2175,6 +2235,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { return () => { closed = true; + setConnectedScopeKey((current) => current === socketScopeKey ? null : current); window.clearTimeout(connectTimer); clearReconnect(); stopPolling(); @@ -2191,11 +2252,16 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { pushToast, canConnectSocket, socketAuthKey, + socketScopeKey, ]); return ( - - {children} - + + + {children} + + ); } diff --git a/ui/src/lib/live-runs-cache.test.ts b/ui/src/lib/live-runs-cache.test.ts index fc388e1ee24..c908e44c590 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 { patchRunStatusInList, removeRunFromList } from "./live-runs-cache"; +import { liveRunRefetchInterval, patchRunStatusInList, removeRunFromList } from "./live-runs-cache"; function run(id: string, status: string): LiveRunForIssue { return { @@ -62,3 +62,19 @@ describe("patchRunStatusInList", () => { expect(next).toBeUndefined(); }); }); + +describe("liveRunRefetchInterval", () => { + it.each(["pending", "error"] as const)("keeps polling %s queries with retained empty data", (status) => { + expect(liveRunRefetchInterval(true, { status, data: [] }, 1000)).toBe(1000); + expect(liveRunRefetchInterval(true, { status, data: null }, 3000)).toBe(3000); + }); + it.each([null, []])("stops known-empty polling with a connected company for %j", (data) => { + expect(liveRunRefetchInterval(true, { status: "success", data }, 1000)).toBe(false); + 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) => { + expect(liveRunRefetchInterval(true, { status: "success", data }, 3000)).toBe(3000); + expect(liveRunRefetchInterval(false, { status: "success", data }, 3000)).toBe(3000); + }); +}); diff --git a/ui/src/lib/live-runs-cache.ts b/ui/src/lib/live-runs-cache.ts index f0533d20c6d..77b09ba29de 100644 --- a/ui/src/lib/live-runs-cache.ts +++ b/ui/src/lib/live-runs-cache.ts @@ -47,3 +47,13 @@ export function patchRunStatusInList( // or its status already matched) so redundant events don't trigger re-renders. return { next: changed ? next : runs, present }; } + +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; +} diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index abac74816d5..7394e3180c3 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -1,3 +1,5 @@ +import { useTaskRunLiveUpdates } from "../context/LiveUpdatesProvider"; +import { liveRunRefetchInterval } from "../lib/live-runs-cache"; import { useTaskRetry } from "../hooks/useTaskRetry"; import { grokBotApi } from "@/api/grokBot"; import { WorkspaceBaseRefRecoveryProvider } from "../components/WorkspaceBaseRefRecovery"; @@ -1204,6 +1206,7 @@ type IssueDetailChatTabProps = { interruption?: Issue["interruption"]; retryPreparing?: boolean; queryIssueId?: string; + hasLiveConnection: boolean; browsers?: import("@paperclipai/shared").TaskBrowser[]; onOpenBrowser?: (browserId: string) => void; onOpenSkill?: (skillId: string, name: string) => void; @@ -1365,6 +1368,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ interruption, retryPreparing, queryIssueId, + hasLiveConnection, browsers, onOpenBrowser, onOpenSkill, @@ -1498,7 +1502,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ queryKey: queryKeys.issues.liveRuns(issueQueryRef), queryFn: () => heartbeatsApi.liveRunsForIssue(issueQueryRef), enabled: !!issueId, - refetchInterval: 1000, + refetchInterval: (query) => liveRunRefetchInterval(hasLiveConnection, query.state, 1000), placeholderData: keepPreviousDataForSameQueryTail(issueQueryRef), }); @@ -1515,7 +1519,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ queryKey: queryKeys.issues.activeRun(issueQueryRef), queryFn: () => heartbeatsApi.activeRunForIssue(issueQueryRef), enabled: activeRunQueryEnabled, - refetchInterval: liveRunCount > 0 ? false : 1000, + refetchInterval: (query) => liveRunCount > 0 ? false : liveRunRefetchInterval(hasLiveConnection, query.state, 1000), placeholderData: keepPreviousDataForSameQueryTail( issueQueryRef, ), @@ -3275,6 +3279,7 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS !hasLegacyIssueDetailQuery(location.search) )); const resolvedCompanyId = issue?.companyId ?? selectedCompanyId; + const hasLiveConnection = useTaskRunLiveUpdates({ companyId: resolvedCompanyId, issueId: issue?.id ?? issueId, issueQueryRef: issueId ?? "pending", assigneeAgentId: issue?.assigneeAgentId }); const externalObjectsState = useIssueExternalObjects(conversation && !conversation.issue ? null : issue?.id ?? null); // A closed isolated workspace no longer blocks the composer. The server reopens // the workspace when the next comment or resume arrives, so the composer stays @@ -3414,7 +3419,7 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS queryKey: queryKeys.issues.liveRuns(issueId!), queryFn: () => heartbeatsApi.liveRunsForIssue(issueId!), enabled: !!issueId, - refetchInterval: 3000, + refetchInterval: (query) => liveRunRefetchInterval(hasLiveConnection, query.state, 3000), select: (runs) => runs.length, placeholderData: keepPreviousDataForSameQueryTail( issueId ?? "pending", @@ -3431,7 +3436,7 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS queryFn: () => heartbeatsApi.activeRunForIssue(issueId!), enabled: !!issueId && (!!issue?.executionRunId || issue?.status === "in_progress"), - refetchInterval: liveRunCount > 0 ? false : 3000, + refetchInterval: (query) => liveRunCount > 0 ? false : liveRunRefetchInterval(hasLiveConnection, query.state, 3000), select: (run) => !!run, placeholderData: keepPreviousDataForSameQueryTail( issueId ?? "pending", @@ -8150,6 +8155,7 @@ export function TaskDetailSurface({ conversation, tasksTab }: { tasksTab?: TaskS interruption={issue.interruption} retryPreparing={taskRetry.optimistic} queryIssueId={issueId ?? issue.id} + hasLiveConnection={hasLiveConnection} onOpenSkill={handleOpenSkill} browsers={browserQuery.data} onOpenBrowser={(id) => { setOpenBrowserId(id); if (isMobile) setMobilePropsOpen(true); else openTaskSidePanel(); }}