Files
PaperClipAI/ui/src/context/LiveUpdatesProvider.recovery.test.tsx
T
Nicky LeachandPaperclip 35ab90ad0f Map runtime activity to authorized task plugin panels (#18131)
## Thinking Path

> - Paperclip manages agents and tasks.
> - Plugins contribute views to the task sidebar.
> - A ready view can belong to an earlier task.
> - Runner records activity under its authenticated run.
> - This PR maps that activity to installed task-panel declarations.
> - The sidebar checks readiness after current-run use and restores use
after reconnect.

## Linked Issues or Issue Description

Builds on merged contract #18129 and Runner publisher #18050. This PR
targets
master and is ready for review. Earlier combined verification is
retained in
[the verification
archive](https://github.com/paperclipai/paperclip/pull/18050#issuecomment-6489369676).

**Subsystem affected**

Run-event reads, plugin task contributions and the task sidebar.

**Problem or motivation**

Readiness polling can show a view left ready by another task. Reload and
reconnect must recover current-run use without losing concurrent live
activity.

**Proposed solution**

Map recorded activity to installed task-panel declarations. Merge
authorized
history and live activity, then perform bounded readiness checks for the
running agent. Preserve the existing read-only view.

## What Changed

- Add optional `{ namespace, activityKey }` activation mappings to
task-panel
declarations. Legacy declarations remain valid but contribute no task
panel.
- Add `view=plugin-panels` to the existing run-events route. Preserve
run-read and
telemetry-read authorization, secret redaction and private-run denial.
Map
activity to installed contributions and return at most 128 panel
identities.
- Scope activation caches to current panel versions and activity
declarations.
Preserve unchanged panels during unrelated plugin updates and history
reads.
- Merge history and live hints without losing concurrent activations.
Filter by
  company, task, run and agent. Restore history on mount and reconnect.
- Replace periodic availability checks with up to three readiness
attempts after
activity. Fresh activity restarts settled transient failures. Pending
checks,
  successful readiness and HTTP 4xx failures are left alone.
- Reuse live run updates with the panel cache key. Recover only panel
history
after a pending subscription read. Mark expected not-ready query results
so
  the global error reporter skips them; real failures remain reportable.
- Preserve current-run dismissal and the read-only plugin view. Update
manifest
  comments and producer/consumer documentation.

## Verification

- Current head: `5e04d757c`, based on master `09c05c550`. Range
comparison
confirmed all six feature commits were unchanged by the rebase. The
final
follow-up keys readiness by the plugin installation ID and availability
path.
- `cd ui && node ../node_modules/vitest/vitest.mjs run
src/hooks/useTaskPluginPanels.test.tsx`
— all 45 cases passed. `cd ui && node ../node_modules/typescript/bin/tsc
-b` passed.
- The shared query-client and live-update recovery selection passed 96
cases on
  `49c544cd2`. Token gates passed before the final hook-only changes.
- Earlier focused verification passed 57 manifest cases, 69
database/history cases,
event-registry cases and route authorization cases, including
private-run denial.
- Regression coverage includes subscription/history races, task changes
during
recovery, unfocused run completion, bounded readiness retries, denied
reads,
expected not-ready responses, changed activity mappings, unchanged panel
preservation, old cache scopes, and same-version route or installation
changes.
- [Current-head
CI](https://github.com/paperclipai/paperclip/actions/runs/40942001707)
passed: 54 successful checks, two skips, no failures or pending checks.
- [Standard and final
review](https://github.com/paperclipai/paperclip/pull/18131#issuecomment-6489842168)
both scored 5/5 on `5e04d757c`. The final review used the Apex tier,
added no
comments, and left no unresolved findings. The PR is conflict-free and
ready for review.
- Local checks stay targeted. GitHub CI provides the broad build, test,
typecheck,
browser and Runner coverage. No deployment or live-provider acceptance
is claimed.

<details>
<summary>Retained failures and recovery evidence</summary>

- `ac9329548` twice failed a browser reload assertion while conversation
history
was concealed. Master fixed its readiness wait in #17923. The approve
journey
then passed locally on `fea63a418` with installed Chrome (1 passed, 39
seconds).
All eight browser shards passed in later CI. The browser test is outside
this diff.
- A canary checkout failed with Git HTTP 408; its unchanged-head retry
passed.
- `de07d09fa` failed the mobile Cases card click after a ten-second
stability wait.
The same read-only journey passed locally on that exact head (1 passed,
26 seconds).
Its unchanged-head CI retry passed 22 cases, with one skip. The original
artifact
  is retained; the cause of that click timeout is not established.
- [CI on
`49c544cd2`](https://github.com/paperclipai/paperclip/actions/runs/40903679075)
passed 54 checks with two skips. CI also passed on `de07d09fa` after the
recorded retry.
- The `1e59ab441` merge checkout passed all browser and server suites.
Runner
shard 2 passed 2,252 cases and timed out waiting for a
controller-recovery command.
The test had landed on master in #18161 after the former branch base.
Rebasing
aligned the local checkout with that code; Runner shard 2 then passed on
`fcf7380d3`.
The original timeout cause remains unproven; no unrelated timeout was
changed.
- [CI on
`fcf7380d3`](https://github.com/paperclipai/paperclip/actions/runs/40940235193)
recorded runner shutdown signals in typecheck and a workspace shard. A
chat
recovery test observed a null wake deadline, and one Runner transport
fixture
failed during close. Those separate assertion failures remain
unexplained.
All of those jobs passed on the final readiness-key head. Those later
passes
establish recovery; they do not prove the cause of the earlier assertion
failures.
- Local browser setup first stopped at pnpm version-download
verification, then a
sandbox IPC restriction, then a missing bundled browser. Installed pnpm
9.15.4
  and Chrome completed the journeys. No dependency files were changed.
- Earlier optional-property TypeScript errors were corrected. Embedded
PostgreSQL
needed its install hook run from its package directory. Older CI also
recorded
fixture-cleanup lock timeouts, attention-fixture timeouts, and canceled
server
shards after runner shutdown. The parent fixed its FIFO snapshot-copy
failure
before merge. The linked verification archive retains earlier combined
evidence.

</details>

## Risks

Activity is a hint and grants no access. Missing plugins, denied reads
and
unavailable views contribute no panel. Older runs without activity
records show
no task plugin tab. A used view stays available for the running task;
reconnect
reconciles readiness. Matching producer and consumer builds are
required. No
migration, dependency, image publication or deployment is included.

## Model Used

OpenAI Codex, based on GPT-6, with reasoning, repository tools and code
execution.
The exact serving snapshot is not exposed.

## 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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-29 13:56:48 -07:00

570 lines
29 KiB
TypeScript

// @vitest-environment jsdom
import { act } from "react";
import { createRoot, type Root } from "react-dom/client";
import { QueryClient, QueryClientProvider, useQuery } from "@tanstack/react-query";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { queryKeys } from "../lib/queryKeys";
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: companySelection.id, selectedCompany: { id: companySelection.id } }),
}));
vi.mock("./ToastContext", () => ({ useToastActions: () => ({ pushToast }) }));
vi.mock("../lib/router", () => ({ useLocation: () => routeSelection }));
vi.mock("../api/auth", () => ({
authApi: { getSession: sessionRead },
}));
vi.mock("../api/health", () => ({ healthApi: { get: healthRead } }));
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;
class Socket {
static instances: Socket[] = [];
readyState = 1;
onopen: (() => void) | null = null;
onclose: (() => void) | null = null;
onmessage: ((message: { data: string }) => void) | null = null;
onerror: (() => void) | null = null;
sent: Array<{ type: string; revision?: number }> = [];
constructor() { Socket.instances.push(this); }
send(encoded: string) { this.sent.push(JSON.parse(encoded)); }
close() { this.readyState = 3; }
}
describe("LiveUpdatesProvider connection recovery", () => {
let root: Root;
let container: HTMLDivElement;
let client: QueryClient;
const read = vi.fn(async () => "current task data");
const pollRead = vi.fn<() => Promise<unknown>>();
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");
sessionRead.mockReset().mockResolvedValue({ user: { id: "viewer" }, session: { id: "session", userId: "viewer" } });
healthRead.mockReset().mockResolvedValue({ deploymentMode: "authenticated" });
Socket.instances = [];
client = new QueryClient({ defaultOptions: { queries: { retry: false, gcTime: Infinity, staleTime: Infinity } } });
client.setQueryData(queryKeys.auth.session, { user: { id: "viewer" }, session: { id: "session", userId: "viewer" } });
client.setQueryData(queryKeys.health, { deploymentMode: "authenticated" });
container = document.createElement("div");
root = createRoot(container);
});
afterEach(async () => {
await act(async () => root.unmount());
client.clear();
vi.restoreAllMocks();
vi.unstubAllGlobals();
vi.useRealTimers();
});
async function render() {
function Tasks() {
const { data } = useQuery({ queryKey: ["company-1", "tasks"], queryFn: read });
return <span>{data}</span>;
}
await act(async () => root.render(
<QueryClientProvider client={client}><LiveUpdatesProvider><Tasks /></LiveUpdatesProvider></QueryClientProvider>,
));
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 <span>{connected ? "connected" : "fallback"}</span>;
}
async function renderPolling(companyId = companySelection.id, withProvider = true) {
const child = <IdleRuns companyId={companyId} />;
await act(async () => root.render(
<QueryClientProvider client={client}>
{withProvider ? <LiveUpdatesProvider>{child}</LiveUpdatesProvider> : child}
</QueryClientProvider>,
));
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([false, true])("discovers a run started in the resync gap after acknowledgement (older read pending=%s)", async pending => {
vi.stubGlobal("WebSocket", Socket);
await renderPolling();
const socket = Socket.instances[0];
const deliver = async (type: string, extras: Record<string, unknown> = {}) => {
await act(async () => socket.onmessage?.({ data: JSON.stringify({ type, version: 1, ...extras }) }));
await act(async () => vi.advanceTimersByTimeAsync(1));
};
await act(async () => socket.onopen?.());
await deliver("live.capabilities", { capabilities: ["run-interests"] });
await deliver("live.subscribed", { revision: socket.sent.at(-1)!.revision });
pollRead.mockClear();
let release!: (value: unknown) => void;
if (pending) pollRead.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
await deliver("live.resync", { revision: socket.sent.at(-1)!.revision });
expect(pollRead).toHaveBeenCalledOnce();
const next = [{ id: "new-run", status: "running" }];
pollRead.mockResolvedValue(next);
await deliver("live.subscribed", { revision: socket.sent.at(-1)!.revision });
if (pending) {
expect(pollRead).toHaveBeenCalledOnce();
await act(async () => release([]));
await act(async () => vi.advanceTimersByTimeAsync(1));
}
expect(pollRead).toHaveBeenCalledTimes(2);
expect(client.getQueryData(["company-1", "idle-run-fixture"])).toEqual(next);
});
it.each(["unmount", "disconnect", "company", "session", "hidden"])("discards a queued resync-ack refresh after %s", async change => {
vi.stubGlobal("WebSocket", Socket);
await renderPolling();
const socket = Socket.instances[0];
const deliver = async (type: string, extras: Record<string, unknown> = {}) => {
await act(async () => socket.onmessage?.({ data: JSON.stringify({ type, version: 1, ...extras }) }));
await act(async () => vi.advanceTimersByTimeAsync(1));
};
await act(async () => socket.onopen?.());
await deliver("live.capabilities", { capabilities: ["run-interests"] });
await deliver("live.subscribed", { revision: socket.sent.at(-1)!.revision });
let release!: (value: unknown) => void;
pollRead.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
await deliver("live.resync", { revision: socket.sent.at(-1)!.revision });
await deliver("live.subscribed", { revision: socket.sent.at(-1)!.revision });
if (change === "unmount") await act(async () => root.unmount());
if (change === "disconnect") await act(async () => socket.onerror?.());
if (change === "company") { companySelection.id = "company-2"; await renderPolling(); }
if (change === "session") {
const next = { user: { id: "viewer" }, session: { id: "replacement", userId: "viewer" } };
sessionRead.mockResolvedValue(next);
await act(async () => client.setQueryData(queryKeys.auth.session, next));
await act(async () => vi.advanceTimersByTimeAsync(1));
}
if (change === "hidden") await act(async () => {
vi.spyOn(document, "visibilityState", "get").mockReturnValue("hidden");
document.dispatchEvent(new Event("visibilitychange"));
});
const reads = pollRead.mock.calls.length;
const invalidate = vi.spyOn(client, "invalidateQueries");
await act(async () => release([]));
await act(async () => vi.advanceTimersByTimeAsync(1));
expect(pollRead).toHaveBeenCalledTimes(reads);
expect(invalidate).not.toHaveBeenCalled();
});
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.each(["DEMO-1", "issue-1"])("discovers and finishes runs with cache key %s on an unfocused non-task route", async queryRef => {
vi.stubGlobal("WebSocket", Socket);
vi.spyOn(document, "hasFocus").mockReturnValue(false);
const runRead = vi.fn<() => Promise<unknown>>().mockResolvedValue([]);
const activeRead = vi.fn<() => Promise<unknown>>().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 <span>{JSON.stringify(runs.data)}</span>;
}
await act(async () => root.render(<QueryClientProvider client={client}><LiveUpdatesProvider><TaskRuns /></LiveUpdatesProvider></QueryClientProvider>));
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<string, unknown>, type = "heartbeat.run.queued") => {
await act(async () => socket.onmessage?.({ data: JSON.stringify({ id: 1, companyId,
type, 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);
runRead.mockResolvedValue([]);
activeRead.mockResolvedValue(null);
await deliver("company-1", { issueId: "issue-1", agentId: "agent-1", runId: "new-run", status: "succeeded" }, "heartbeat.run.status");
expect(client.getQueryData(queryKeys.issues.liveRuns(queryRef))).toEqual([]);
expect(container.textContent).not.toContain("new-run");
});
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<unknown>>().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 <span>{JSON.stringify(data)}</span>;
}
await act(async () => root.render(<QueryClientProvider client={client}><LiveUpdatesProvider><TaskRuns /></LiveUpdatesProvider></QueryClientProvider>));
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<unknown>>().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(<QueryClientProvider client={client}><LiveUpdatesProvider><TaskRuns task={task} /></LiveUpdatesProvider></QueryClientProvider>));
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<unknown>>().mockResolvedValue([]);
const activeRead = vi.fn<() => Promise<unknown>>().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 <><Reads connected={connected} interval={1000} /><Reads connected={connected} interval={3000} /></>;
}
await act(async () => root.render(<QueryClientProvider client={client}><LiveUpdatesProvider><TaskRuns /></LiveUpdatesProvider></QueryClientProvider>));
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 <span>{connected ? "connected" : "fallback"}</span>;
}
await act(async () => root.render(<QueryClientProvider client={client}><LiveUpdatesProvider><UnassignedTask /></LiveUpdatesProvider></QueryClientProvider>));
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();
expect(container.textContent).toBe("current task data");
expect(Socket.instances).toHaveLength(1);
// The server saves a reply after the page read but before it subscribes.
// There is no event replay and this is not a reconnect.
read.mockResolvedValue("reply saved during connection setup");
await act(async () => Socket.instances[0].onopen?.());
await act(async () => vi.advanceTimersByTimeAsync(1));
expect(container.textContent).toBe("reply saved during connection setup");
expect(sessionRead).not.toHaveBeenCalled();
expect(healthRead).not.toHaveBeenCalled();
const readsAfterConnect = read.mock.calls.length;
await act(async () => vi.advanceTimersByTimeAsync(30_000));
expect(read).toHaveBeenCalledTimes(readsAfterConnect);
await act(async () => Socket.instances[0].onclose?.());
await act(async () => vi.advanceTimersByTimeAsync(1_001));
await act(async () => Socket.instances.at(-1)?.onopen?.());
await act(async () => vi.advanceTimersByTimeAsync(1));
expect(sessionRead).toHaveBeenCalled();
expect(healthRead).toHaveBeenCalled();
});
it.each(["missing", "throws"])("polls visible data and resumes realtime when the constructor %s", async (failure) => {
vi.stubGlobal("WebSocket", failure === "missing" ? undefined : class {
constructor() { throw new DOMException("Blocked", "SecurityError"); }
});
await render();
expect(container.textContent).toBe("current task data");
expect(read).toHaveBeenCalledTimes(1);
await act(async () => vi.advanceTimersByTimeAsync(15_000));
expect(read).toHaveBeenCalledTimes(2);
vi.stubGlobal("WebSocket", Socket);
await act(async () => vi.advanceTimersByTimeAsync(15_000));
expect(Socket.instances).toHaveLength(1);
await act(async () => Socket.instances[0].onopen?.());
const readsAfterRecovery = read.mock.calls.length;
await act(async () => vi.advanceTimersByTimeAsync(30_000));
expect(read).toHaveBeenCalledTimes(readsAfterRecovery);
expect(Socket.instances).toHaveLength(1);
// An ordinary disconnect must re-enable the same fallback.
await act(async () => Socket.instances[0].onclose?.());
await act(async () => vi.advanceTimersByTimeAsync(15_000));
expect(read.mock.calls.length).toBeGreaterThan(readsAfterRecovery);
});
it("stops fallback reads and connection retries while hidden and after unmount", async () => {
const construct = vi.fn();
vi.stubGlobal("WebSocket", class {
constructor() { construct(); throw new Error("Unavailable"); }
});
await render();
const visibility = vi.spyOn(document, "visibilityState", "get");
await act(async () => {
visibility.mockReturnValue("hidden");
document.dispatchEvent(new Event("visibilitychange"));
});
const attempts = construct.mock.calls.length;
await act(async () => vi.advanceTimersByTimeAsync(30_000));
expect(read).toHaveBeenCalledTimes(1);
expect(construct).toHaveBeenCalledTimes(attempts);
await act(async () => {
visibility.mockReturnValue("visible");
document.dispatchEvent(new Event("visibilitychange"));
});
await act(async () => vi.advanceTimersByTimeAsync(15_001));
expect(read.mock.calls.length).toBeGreaterThan(1);
await act(async () => root.unmount());
const readsAtUnmount = read.mock.calls.length;
const attemptsAtUnmount = construct.mock.calls.length;
await act(async () => vi.advanceTimersByTimeAsync(30_000));
expect(read).toHaveBeenCalledTimes(readsAtUnmount);
expect(construct).toHaveBeenCalledTimes(attemptsAtUnmount);
});
});