mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 22:24:16 +01:00
feat(live): deliver recoverable transcripts to interested views (#17456)
## Thinking Path > - Paperclip should show live agent output without repeating presentation work for every view. > - Execution, acknowledgements and durable checkpoints must remain independent of browser interest. > - The server already owns authorization and redaction at delivery. > - A shared browser connection can declare which visible runs need full transcripts. > - Catch-up must handle subscription races, disconnects, terminal output and slow consumers. > - This PR delivers the backend and UI contract together without a database migration. ## Linked Issues or Issue Description **What existing behavior does this improve?** Visible transcript consumers receive recoverable output through one shared browser connection. **Current behavior** Independent views can create duplicate connections, presentation work and recurring history reads. A subscription race or filtered native history can lose the saved tail. **Proposed behavior** Negotiate run interests, subscribe before catch-up and retain independent canonical recovery cursors. Keep lightweight lifecycle and attention events available without full transcript delivery. PR 4 of #17097 → #17439 → #17455 → #17456. The runtime feature is independent of hot-upgrade recovery; the stack order is for review convenience. ## What Changed - Add additive WebSocket capability negotiation, replacement interests, revision acknowledgement and resynchronization notices. - Reference-count visible consumer interests on one company-scoped browser connection. Preserve older-client/server behavior. - Reuse canonical structural/current-user/registered-secret redaction for plugin run notifications through the same bounded delivery queue. Keep producer and uninterested streams free of recipient lookups. - Refresh application queries after the matching recovery acknowledgement, including one follow-up after an older in-flight read. Ordinary interest acknowledgements add no refresh. - Keep authorization and redaction at delivery. Browser interest cannot change harness execution, PRP ACKs, checkpoints, accounting or canonical logs. - Deduplicate by sequence and log identity. Catch up terminal output and saved tails without recurring reads during healthy negotiated streaming. - Bound delivery to 1,000 items and 1 MiB. Overflow requires resynchronization; persistent congestion closes the connection. - Preserve compact-history expansion from current master across paginated catch-up. Snapshot the original read window before React can defer its update. ## Verification On current source `f6710f0b3c`, Linux [job 122604880503](https://github.com/paperclipai/paperclip/actions/runs/40624101891/job/122604880503) executes all five run-read permission cases successfully (2.574 seconds), including grant and revocation on the same connection. Its shard passes all 2,430 cases. Current [server-shard job 122604880416](https://github.com/paperclipai/paperclip/actions/runs/40624101891/job/122604880416) has 1,513 passes, one failure and three skips: an unchanged nested-clone expiration/restore test exceeds its 15-second deadline. The preceding `5efc6c48bf` source has identical non-document bytes; [its corresponding job 122601359272](https://github.com/paperclipai/paperclip/actions/runs/40623017917/job/122601359272) passes all 21 nested-clone cases, including that case in 9.505 seconds. The exact timing cause is unproven. The single diagnostic [retry job 122609332597](https://github.com/paperclipai/paperclip/actions/runs/40624101891/job/122609332597) passes all 21 nested-clone cases (61.316 seconds), including the formerly timed-out case in 7.003 seconds, and all 1,514 shard cases (three skips). The failure remains recorded; no timeout or assertion was changed. After the server squash merge and qualification retarget, all 50 focused server cases and all 91 focused UI cases pass again, with both typechecks and token gates. Runner vendor preparation rebuilds current master inputs before those tests. This source needs fresh CI and reviews; prior CI and browser receipts do not qualify it by inference. Current candidate `f6710f0b3c` now targets master `6987c2f20b`, after merged #17097, #17439 and #17455. Its diff contains only transcript delivery and its documentation. It adds no database migration. Retargeting changes only the evidence document; all production and test code is byte-identical to `5efc6c48bf`. On the eighth integration, all 50 focused server delivery cases and all 91 focused UI cases pass, with both typechecks and token gates. The following 35-case receipt belongs to the earlier delivery correction. Four low-queue authorization/decoration stall variants and the late-rejection case fail before correction. They now verify the five-second read deadline, old and negotiated clients, revisions that cannot reset the deadline, unaffected sibling connections, rejected late results, and timer cleanup. The first combined toast run has four loopback EPERM failures and 31 passes; its permitted repeat passes all 35. Earlier 28-case evidence excluded the toast file. The permission correction requires the current HTTP company-scope transcript permission, plus task access, for both native events and legacy logs. Prior Linux source `49a8783e9b`, job 122549913643, executes all five run-read cases, including grant/revocation on the same connection, successfully. That source placed this file in server shard 11; later shard assignments can differ and must be checked against their actual inventories. The older `405bedd189` receipt retains a separate scheduled-retry teardown race and inherited fixture collision; neither is waived. The new head must pass again in Linux CI. The seven extracted stream-recovery tests preserve their assertions. All 39 related legacy/native cases, UI typecheck and token gates pass after restoring the initially missed fixture helper. Earlier full workspace build/typecheck and Chrome journeys at `445a76563c` cover two views, company switches, reconnect, terminal catch-up, reload and zero recurring healthy transcript reads. Those browser receipts are historical source evidence and do not qualify this new head by inference. The [qualification record](https://github.com/paperclipai/paperclip/blob/f6710f0b3c84c6fdb2021df67206261b446dfc97/doc/plans/2026-10-22-hot-upgrade-qualification.md) retains exact earlier UI/history, source-boundary and cleanup failures with their fixes and limits. The workload-cost test preserves bounds and authorization callback counts for zero interests, one viewer, multiple browsers and a slow viewer. Its socket congestion is simulated. It measures neither SQL nor network costs. Current permission checks and the final hosted active-agent/viewer matrix remain unmeasured; no production savings are claimed. Fresh exact-head CI, legacy status, completed Greptile 5/5, resolved feedback and ordinary merge eligibility remain required. The earlier recipient corrections pass all 47 registry/plugin/delivery/WebSocket/toast cases and all 174 UI provider/recovery/interest cases, both typechecks and token gates on source `b202e59a25`. A real bus/host-service test registers a synthetic run secret through the production registry and reproduces its raw notification before the fix; native/legacy/final notifications now redact it, disposal prevents late worker notification. A later P1 review reproduces lost final notifications on settings-read and secret-resolution failure. Both cases fail before correction. Any delivery failure or subscription timeout now sends one content-free stream error and removes the plugin subscription. It never changes the agent execution or run status. Later events start no new redaction reads. The final integrated six-file server lane passes all 50 cases, with server typecheck and token gates. Existing browser delivery behavior is unchanged. Storage and secret-provider seams are scripted. Two real React Query/provider regressions first miss a run during resync; both now refresh after the valid replacement ACK, including pending-read and stale-scope cases. The prerequisite qualification batch passes all 118 cases and E2E typecheck after master integration. Fresh CI, security scan and reviews remain pending on this head. The contribution-rule review correction removes 104 explanatory comments introduced by this slice and keeps the ordering, deadline and plugin-delivery explanations in the delivery document. Existing comments and functional test/lint directives remain. All edited source files have identical parsed non-comment tokens before and after the change. All 50 server delivery cases and all 91 cases in six affected UI suites pass, along with token gates. The new head requires fresh CI, security and review. ## Risks A slow or disconnected viewer must resynchronize from authorized canonical history. It may see a display gap until catch-up completes. Company changes and revoked access must remove delivery interests. This slice adds no database migration and cannot change execution or recovery authority. Full transcript delivery evaluates current company-scope permission per delivery. Active plugin recipients also perform canonical redaction reads; viewer-load database costs on this final source remain unmeasured. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools and code execution. The runtime does not expose a more specific model ID or context-window size. ## 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 - [ ] 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>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
f52da40f08
commit
6cc765bd81
40 files changed
+3610
-1092
No files matched your search
@@ -0,0 +1,321 @@
|
||||
// @vitest-environment jsdom
|
||||
|
||||
import { act, type ReactNode } from "react";
|
||||
import { createRoot as createReactRoot } from "react-dom/client";
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { __liveUpdatesTestUtils, type CompanyLiveEventHandler } from "../../context/LiveUpdatesProvider";
|
||||
import { LiveTranscriptInterests } from "../../lib/live-transcript-interests";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { useLiveRunTranscripts } from "./useLiveRunTranscripts";
|
||||
|
||||
const { useQueryMock, logMock, buildTranscriptMock } = vi.hoisted(() => ({
|
||||
useQueryMock: vi.fn(() => ({ data: { censorUsernameInLogs: false } })),
|
||||
logMock: vi.fn(async (_runId: string, _offset: number, _limit: number, _options: { signal: AbortSignal }) =>
|
||||
({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 0 })),
|
||||
buildTranscriptMock: vi.fn((chunks: unknown[]) => chunks),
|
||||
}));
|
||||
|
||||
vi.mock("@tanstack/react-query", () => ({
|
||||
useQuery: useQueryMock,
|
||||
}));
|
||||
|
||||
vi.mock("../../api/instanceSettings", () => ({
|
||||
instanceSettingsApi: {
|
||||
getGeneral: vi.fn(),
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("../../api/heartbeats", () => ({
|
||||
heartbeatsApi: {
|
||||
log: logMock,
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("../../adapters", () => ({
|
||||
buildTranscript: buildTranscriptMock,
|
||||
getUIAdapter: () => null,
|
||||
onAdapterChange: () => () => {},
|
||||
}));
|
||||
|
||||
const { LiveEventSubscriptionContext } = __liveUpdatesTestUtils;
|
||||
let listeners: Set<CompanyLiveEventHandler>;
|
||||
let interests: LiveTranscriptInterests;
|
||||
let sent: Array<{ type: string; revision?: number; runIds?: string[] }>;
|
||||
let socketConstructed: ReturnType<typeof vi.fn<() => void>>;
|
||||
function createRoot(container: HTMLDivElement) {
|
||||
const root = createReactRoot(container);
|
||||
const subscription = {
|
||||
companyId: "company-1", transcripts: interests,
|
||||
subscribe(handler: CompanyLiveEventHandler) {
|
||||
listeners.add(handler);
|
||||
return () => { listeners.delete(handler); };
|
||||
},
|
||||
};
|
||||
return {
|
||||
unmount: () => root.unmount(),
|
||||
render: (children: ReactNode) => root.render(
|
||||
<LiveEventSubscriptionContext.Provider value={subscription}>{children}</LiveEventSubscriptionContext.Provider>,
|
||||
),
|
||||
};
|
||||
}
|
||||
function sendEvent(event: Omit<LiveEvent, "id">) {
|
||||
for (const listener of listeners) listener({ ...event, id: 1 });
|
||||
}
|
||||
function negotiate() {
|
||||
interests.open({ send: data => sent.push(JSON.parse(data)) });
|
||||
interests.receive({ type: "live.capabilities", version: 1, capabilities: ["run-interests"] });
|
||||
acknowledge();
|
||||
}
|
||||
function acknowledge() {
|
||||
interests.receive({ type: "live.subscribed", version: 1, revision: sent.at(-1)?.revision });
|
||||
}
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
(globalThis as any).IS_REACT_ACT_ENVIRONMENT = true;
|
||||
|
||||
describe("useLiveRunTranscripts shared-stream recovery", () => {
|
||||
const OriginalWebSocket = globalThis.WebSocket;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible");
|
||||
listeners = new Set(); interests = new LiveTranscriptInterests(); sent = [];
|
||||
socketConstructed = vi.fn();
|
||||
useQueryMock.mockClear();
|
||||
logMock.mockReset();
|
||||
logMock.mockImplementation(async (_id, offset) => ({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: offset }));
|
||||
buildTranscriptMock.mockClear();
|
||||
globalThis.WebSocket = class { constructor() { socketConstructed(); } } as unknown as typeof WebSocket;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
globalThis.WebSocket = OriginalWebSocket;
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("fetches a consecutive truncated row excluded from an in-flight history response", async () => {
|
||||
vi.useFakeTimers();
|
||||
let value: ReturnType<typeof useLiveRunTranscripts>;
|
||||
function Harness() {
|
||||
value = useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local" }] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
const row = (seq: number, chunk: string) => JSON.stringify({ seq, ts: "2026-10-24T00:00:00Z", stream: "stdout", chunk }) + "\n";
|
||||
const first = row(1, "first");
|
||||
const second = row(2, "complete second output");
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
let finish!: (value: Awaited<ReturnType<typeof logMock>>) => void;
|
||||
logMock.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
|
||||
await act(async () => negotiate());
|
||||
await act(async () => {
|
||||
sendEvent({ companyId: "company-1", type: "heartbeat.run.log", createdAt: "2026-10-24T00:00:00Z",
|
||||
payload: { runId: "run-1", seq: 1, chunk: "first", stream: "stdout" } });
|
||||
sendEvent({ companyId: "company-1", type: "heartbeat.run.log", createdAt: "2026-10-24T00:00:00Z",
|
||||
payload: { runId: "run-1", seq: 2, chunk: "short", truncated: true, stream: "stdout" } });
|
||||
});
|
||||
await act(async () => finish({ runId: "run-1", store: "memory", logRef: "log-1", content: first, nextOffset: first.length }));
|
||||
const beforeRetry = logMock.mock.calls.length;
|
||||
logMock.mockResolvedValueOnce({ runId: "run-1", store: "memory", logRef: "log-1", content: second, nextOffset: first.length + second.length });
|
||||
await act(async () => vi.advanceTimersByTimeAsync(2000));
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(beforeRetry);
|
||||
expect(value!.transcriptByRun.get("run-1")).toEqual(expect.arrayContaining([expect.objectContaining({ chunk: "complete second output", seq: 2 })]));
|
||||
const settledReads = logMock.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(logMock).toHaveBeenCalledTimes(settledReads);
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
|
||||
it("finishes slow legacy pages while a sibling catches up repeated truncated output", async () => {
|
||||
vi.useFakeTimers();
|
||||
let value: ReturnType<typeof useLiveRunTranscripts>;
|
||||
function Harness() {
|
||||
value = useLiveRunTranscripts({ companyId: "company-1", runs: [
|
||||
{ id: "run-1", status: "running", adapterType: "codex_local" },
|
||||
{ id: "slow", status: "succeeded", adapterType: "codex_local" },
|
||||
] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
type Result = Awaited<ReturnType<typeof logMock>>;
|
||||
const slow: Array<{ resolve: (value: Result) => void; signal: AbortSignal }> = [];
|
||||
const row = (seq: number, chunk: string) => JSON.stringify({ seq, ts: "2026-10-24T00:00:00Z", stream: "stdout", chunk }) + "\n";
|
||||
let seq = 0;
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
logMock.mockImplementation(async (id, offset, _limit, { signal }) => {
|
||||
if (id === "slow") return new Promise(resolve => slow.push({ resolve, signal }));
|
||||
return { runId: id, store: "memory", logRef: "log-1", content: offset < seq ? row(seq, `complete ${seq}`) : "", nextOffset: seq };
|
||||
});
|
||||
await act(async () => negotiate());
|
||||
for (seq = 1; seq <= 3; seq++) {
|
||||
await act(async () => sendEvent({ companyId: "company-1", type: "heartbeat.run.log", createdAt: "2026-10-24T00:00:00Z",
|
||||
payload: { runId: "run-1", seq, truncated: true } }));
|
||||
}
|
||||
expect(slow).toHaveLength(1);
|
||||
expect(slow[0].signal.aborted).toBe(false);
|
||||
expect(value!.transcriptByRun.get("run-1")).toEqual(expect.arrayContaining([expect.objectContaining({ seq: 3, chunk: "complete 3" })]));
|
||||
await act(async () => slow[0].resolve({ runId: "slow", store: "memory", logRef: "log-1", content: row(1, "older output"), nextOffset: 1 }));
|
||||
expect(slow).toHaveLength(2);
|
||||
expect(slow[1].signal.aborted).toBe(false);
|
||||
await act(async () => slow[1].resolve({ runId: "slow", store: "memory", logRef: "log-1", content: "", nextOffset: 1 }));
|
||||
expect(value!.transcriptByRun.get("slow")).toEqual(expect.arrayContaining([expect.objectContaining({ chunk: "older output" })]));
|
||||
const reads = logMock.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(logMock).toHaveBeenCalledTimes(reads);
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
|
||||
it("coalesces legacy terminal catch-up behind an in-flight read without aborting it", async () => {
|
||||
let value: ReturnType<typeof useLiveRunTranscripts>;
|
||||
function Harness() {
|
||||
value = useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local" }] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
let finish!: (result: Awaited<ReturnType<typeof logMock>>) => void;
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
logMock.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
|
||||
await act(async () => negotiate());
|
||||
const reads = logMock.mock.calls.length;
|
||||
const signal = logMock.mock.calls.at(-1)![3].signal;
|
||||
await act(async () => {
|
||||
for (let i = 0; i < 3; i++) sendEvent({ companyId: "company-1", type: "heartbeat.run.status", createdAt: "2026-10-24T00:00:00Z",
|
||||
payload: { runId: "run-1", status: "succeeded" } });
|
||||
});
|
||||
expect(signal.aborted).toBe(false);
|
||||
expect(logMock).toHaveBeenCalledTimes(reads);
|
||||
logMock.mockResolvedValueOnce({ runId: "run-1", store: "memory", logRef: "log-1", content:
|
||||
JSON.stringify({ seq: 1, ts: "2026-10-24T00:00:00Z", stream: "stdout", chunk: "terminal output" }) + "\n", nextOffset: 1 });
|
||||
await act(async () => finish({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 0 }));
|
||||
expect(logMock).toHaveBeenCalledTimes(reads + 2);
|
||||
expect(value!.transcriptByRun.get("run-1")).toEqual(expect.arrayContaining([expect.objectContaining({ chunk: "terminal output" })]));
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
|
||||
it("keeps a legacy catch-up in flight through unrelated interest replacements", async () => {
|
||||
vi.useFakeTimers();
|
||||
let value: ReturnType<typeof useLiveRunTranscripts>;
|
||||
function Harness() {
|
||||
value = useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local" }] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
let finish!: (result: Awaited<ReturnType<typeof logMock>>) => void;
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
logMock.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
|
||||
await act(async () => negotiate());
|
||||
const reads = logMock.mock.calls.length;
|
||||
const signal = logMock.mock.calls.at(-1)![3].signal;
|
||||
for (let index = 0; index < 4; index++) {
|
||||
let release!: () => void;
|
||||
await act(async () => { release = interests.acquire([`native-sibling-${index}`]); });
|
||||
expect(signal.aborted).toBe(false);
|
||||
await act(async () => acknowledge());
|
||||
await act(async () => release());
|
||||
await act(async () => acknowledge());
|
||||
expect(signal.aborted).toBe(false);
|
||||
expect(logMock).toHaveBeenCalledTimes(reads);
|
||||
}
|
||||
await act(async () => finish({ runId: "run-1", store: "memory", logRef: "log-1", content:
|
||||
JSON.stringify({ seq: 1, ts: "2026-10-24T00:00:00Z", stream: "stdout", chunk: "uninterrupted legacy output" }) + "\n", nextOffset: 1 }));
|
||||
expect(value!.transcriptByRun.get("run-1")).toEqual(expect.arrayContaining([expect.objectContaining({ chunk: "uninterrupted legacy output" })]));
|
||||
const settled = logMock.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(logMock).toHaveBeenCalledTimes(settled);
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
|
||||
it("keeps HTTP polling until interest acknowledgement, then stops recurring reads", async () => {
|
||||
vi.useFakeTimers();
|
||||
function Harness({ bytes = 0 }: { bytes?: number }) {
|
||||
useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local", lastOutputBytes: bytes }] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(30_000));
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(1);
|
||||
await act(async () => negotiate());
|
||||
const afterCatchup = logMock.mock.calls.length;
|
||||
await act(async () => root.render(<Harness bytes={512} />));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(logMock).toHaveBeenCalledTimes(afterCatchup);
|
||||
await act(async () => interests.close());
|
||||
const afterDisconnect = logMock.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(30_000));
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(afterDisconnect);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
|
||||
it("reads past a running run's stale byte count after hide/show without another live event", async () => {
|
||||
vi.useFakeTimers();
|
||||
const visibility = vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible");
|
||||
const line = (seq: number, chunk: string) => JSON.stringify({ seq, chunk, ts: "2026-10-26T00:00:00Z", stream: "stdout" }) + "\n";
|
||||
const segments = [line(1, "initial output")];
|
||||
const originalBytes = segments[0]!.length;
|
||||
logMock.mockImplementation(async (_id, offset) => {
|
||||
let cursor = 0;
|
||||
for (const segment of segments) {
|
||||
if (cursor === offset) return { runId: "run-1", store: "memory", logRef: "log-1", content: segment, nextOffset: offset + segment.length };
|
||||
cursor += segment.length;
|
||||
}
|
||||
return { runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: offset };
|
||||
});
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local", logBytes: originalBytes }] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
await act(async () => negotiate());
|
||||
await act(async () => { visibility.mockReturnValue("hidden"); document.dispatchEvent(new Event("visibilitychange")); });
|
||||
segments.push(line(2, "saved during disconnect"), line(3, "last saved page"));
|
||||
await act(async () => { visibility.mockReturnValue("visible"); document.dispatchEvent(new Event("visibilitychange")); });
|
||||
await act(async () => acknowledge());
|
||||
const chunks = buildTranscriptMock.mock.calls.at(-1)?.[0];
|
||||
expect(chunks).toEqual([
|
||||
expect.objectContaining({ chunk: "initial output" }),
|
||||
expect.objectContaining({ chunk: "saved during disconnect" }),
|
||||
expect.objectContaining({ chunk: "last saved page" }),
|
||||
]);
|
||||
const caughtUp = logMock.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(logMock).toHaveBeenCalledTimes(caughtUp);
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
|
||||
it("catches up after resynchronization, hide/show, and terminal status", async () => {
|
||||
const visibility = vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible");
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local" }] });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
await act(async () => negotiate());
|
||||
const prior = logMock.mock.calls.length;
|
||||
await act(async () => { interests.receive({ type: "live.resync", version: 1, revision: sent.at(-1)?.revision }); acknowledge(); });
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(prior);
|
||||
await act(async () => { visibility.mockReturnValue("hidden"); document.dispatchEvent(new Event("visibilitychange")); });
|
||||
expect(sent.at(-1)?.runIds).toEqual([]);
|
||||
const hiddenReads = logMock.mock.calls.length;
|
||||
await act(async () => acknowledge());
|
||||
expect(logMock).toHaveBeenCalledTimes(hiddenReads);
|
||||
await act(async () => { visibility.mockReturnValue("visible"); document.dispatchEvent(new Event("visibilitychange")); });
|
||||
await act(async () => acknowledge());
|
||||
expect(sent.at(-1)?.runIds).toEqual(["run-1"]);
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(hiddenReads);
|
||||
const beforeTerminal = logMock.mock.calls.length;
|
||||
await act(async () => sendEvent({ companyId: "company-1", type: "heartbeat.run.status", createdAt: new Date().toISOString(), payload: { runId: "run-1", status: "succeeded" } }));
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(beforeTerminal);
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
});
|
||||
@@ -1,7 +1,10 @@
|
||||
// @vitest-environment jsdom
|
||||
|
||||
import { act } from "react";
|
||||
import { createRoot } from "react-dom/client";
|
||||
import { act, type ReactNode } from "react";
|
||||
import { createRoot as createReactRoot } from "react-dom/client";
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { __liveUpdatesTestUtils, type CompanyLiveEventHandler } from "../../context/LiveUpdatesProvider";
|
||||
import { LiveTranscriptInterests } from "../../lib/live-transcript-interests";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { ApiError } from "../../api/client";
|
||||
import { useLiveRunTranscripts } from "./useLiveRunTranscripts";
|
||||
@@ -9,7 +12,8 @@ import { TRANSCRIPT_REQUEST_TIMEOUT_MS } from "./read-transcript-request";
|
||||
|
||||
const { useQueryMock, logMock, buildTranscriptMock } = vi.hoisted(() => ({
|
||||
useQueryMock: vi.fn(() => ({ data: { censorUsernameInLogs: false } })),
|
||||
logMock: vi.fn(async () => ({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 0 })),
|
||||
logMock: vi.fn(async (_runId: string, _offset: number, _limit: number, _options: { signal: AbortSignal }) =>
|
||||
({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 0 })),
|
||||
buildTranscriptMock: vi.fn((chunks: unknown[]) => chunks),
|
||||
}));
|
||||
|
||||
@@ -35,40 +39,37 @@ vi.mock("../../adapters", () => ({
|
||||
onAdapterChange: () => () => {},
|
||||
}));
|
||||
|
||||
class FakeWebSocket {
|
||||
static readonly CONNECTING = 0;
|
||||
static readonly OPEN = 1;
|
||||
static readonly CLOSING = 2;
|
||||
static readonly CLOSED = 3;
|
||||
static instances: FakeWebSocket[] = [];
|
||||
|
||||
readonly url: string;
|
||||
readyState = FakeWebSocket.CONNECTING;
|
||||
onopen: ((event: Event) => void) | null = null;
|
||||
onmessage: ((event: MessageEvent) => void) | null = null;
|
||||
onerror: ((event: Event) => void) | null = null;
|
||||
onclose: ((event: CloseEvent) => void) | null = null;
|
||||
closeCalls: Array<{ code?: number; reason?: string }> = [];
|
||||
|
||||
constructor(url: string) {
|
||||
this.url = url;
|
||||
FakeWebSocket.instances.push(this);
|
||||
}
|
||||
|
||||
close(code?: number, reason?: string) {
|
||||
this.closeCalls.push({ code, reason });
|
||||
this.readyState = FakeWebSocket.CLOSING;
|
||||
}
|
||||
|
||||
triggerOpen() {
|
||||
this.readyState = FakeWebSocket.OPEN;
|
||||
this.onopen?.(new Event("open"));
|
||||
}
|
||||
|
||||
triggerClose() {
|
||||
this.readyState = FakeWebSocket.CLOSED;
|
||||
this.onclose?.(new CloseEvent("close"));
|
||||
}
|
||||
const { LiveEventSubscriptionContext } = __liveUpdatesTestUtils;
|
||||
let listeners: Set<CompanyLiveEventHandler>;
|
||||
let interests: LiveTranscriptInterests;
|
||||
let sent: Array<{ type: string; revision?: number; runIds?: string[] }>;
|
||||
let socketConstructed: ReturnType<typeof vi.fn<() => void>>;
|
||||
function createRoot(container: HTMLDivElement) {
|
||||
const root = createReactRoot(container);
|
||||
const subscription = {
|
||||
companyId: "company-1", transcripts: interests,
|
||||
subscribe(handler: CompanyLiveEventHandler) {
|
||||
listeners.add(handler);
|
||||
return () => { listeners.delete(handler); };
|
||||
},
|
||||
};
|
||||
return {
|
||||
unmount: () => root.unmount(),
|
||||
render: (children: ReactNode) => root.render(
|
||||
<LiveEventSubscriptionContext.Provider value={subscription}>{children}</LiveEventSubscriptionContext.Provider>,
|
||||
),
|
||||
};
|
||||
}
|
||||
function sendEvent(event: Omit<LiveEvent, "id">) {
|
||||
for (const listener of listeners) listener({ ...event, id: 1 });
|
||||
}
|
||||
function negotiate() {
|
||||
interests.open({ send: data => sent.push(JSON.parse(data)) });
|
||||
interests.receive({ type: "live.capabilities", version: 1, capabilities: ["run-interests"] });
|
||||
acknowledge();
|
||||
}
|
||||
function acknowledge() {
|
||||
interests.receive({ type: "live.subscribed", version: 1, revision: sent.at(-1)?.revision });
|
||||
}
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
@@ -79,12 +80,13 @@ describe("useLiveRunTranscripts", () => {
|
||||
|
||||
beforeEach(() => {
|
||||
vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible");
|
||||
FakeWebSocket.instances = [];
|
||||
listeners = new Set(); interests = new LiveTranscriptInterests(); sent = [];
|
||||
socketConstructed = vi.fn();
|
||||
useQueryMock.mockClear();
|
||||
logMock.mockReset();
|
||||
logMock.mockImplementation(async () => ({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: 0 }));
|
||||
logMock.mockImplementation(async (_id, offset) => ({ runId: "run-1", store: "memory", logRef: "log-1", content: "", nextOffset: offset }));
|
||||
buildTranscriptMock.mockClear();
|
||||
globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket;
|
||||
globalThis.WebSocket = class { constructor() { socketConstructed(); } } as unknown as typeof WebSocket;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
@@ -108,7 +110,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
await act(async () => root.render(<Harness />));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(10_000));
|
||||
expect(logMock).not.toHaveBeenCalled();
|
||||
expect(FakeWebSocket.instances).toHaveLength(0);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
await act(async () => {
|
||||
visibility.mockReturnValue("visible");
|
||||
document.dispatchEvent(new Event("visibilitychange"));
|
||||
@@ -136,6 +138,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
const result = (content: string) => ({ runId: "run-1", store: "memory", logRef: "log-1", content, nextOffset: 100_000 });
|
||||
let resolveExpanded!: (value: ReturnType<typeof result>) => void;
|
||||
logMock.mockResolvedValueOnce(result(`${row(2, "Latest answer")}\n`));
|
||||
logMock.mockResolvedValueOnce(result(""));
|
||||
logMock.mockImplementationOnce(() => new Promise(resolve => { resolveExpanded = resolve; }));
|
||||
function Harness({ limit, bytes = 100_000 }: { limit: number; bytes?: number }) {
|
||||
({ truncatedRunIds } = useLiveRunTranscripts({
|
||||
@@ -149,7 +152,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
const chunks = () => buildTranscriptMock.mock.calls.at(-1)?.[0];
|
||||
try {
|
||||
await act(async () => root.render(<Harness limit={32_000} />));
|
||||
expect(logMock).toHaveBeenLastCalledWith("run-1", 68_000, 32_000, expect.objectContaining({ alignStartToRecord: true }));
|
||||
expect(logMock).toHaveBeenNthCalledWith(1, "run-1", 68_000, 32_000, expect.objectContaining({ alignStartToRecord: true }));
|
||||
expect(truncatedRunIds.has("run-1")).toBe(true);
|
||||
expect(chunks()).toEqual([expect.objectContaining({ chunk: "Latest answer" })]);
|
||||
await act(async () => root.render(<Harness limit={256_000} />));
|
||||
@@ -173,7 +176,9 @@ describe("useLiveRunTranscripts", () => {
|
||||
const row = (seq: number) => `${JSON.stringify({ seq, chunk: `Output ${seq}`, stream: "stdout", ts: "2026-10-24T00:00:00Z" })}\n`;
|
||||
const response = { runId: "run-1", store: "memory", logRef: "log-1" };
|
||||
logMock.mockResolvedValueOnce({ ...response, content: row(1), nextOffset: 100 });
|
||||
logMock.mockResolvedValueOnce({ ...response, content: "", nextOffset: 100 });
|
||||
logMock.mockResolvedValueOnce({ ...response, content: row(2), nextOffset: 100_000 });
|
||||
logMock.mockResolvedValueOnce({ ...response, content: "", nextOffset: 100_000 });
|
||||
function Harness({ bytes }: { bytes: number }) {
|
||||
({ truncatedRunIds } = useLiveRunTranscripts({
|
||||
runs: [{ id: "run-1", status: "succeeded", adapterType: "codex_local", logBytes: bytes }],
|
||||
@@ -184,10 +189,10 @@ describe("useLiveRunTranscripts", () => {
|
||||
const root = createRoot(document.createElement("div"));
|
||||
try {
|
||||
await act(async () => root.render(<Harness bytes={100} />));
|
||||
expect(logMock).toHaveBeenLastCalledWith("run-1", 0, 32_000, expect.anything());
|
||||
expect(logMock).toHaveBeenNthCalledWith(1, "run-1", 0, 32_000, expect.anything());
|
||||
expect(truncatedRunIds.has("run-1")).toBe(false);
|
||||
await act(async () => root.render(<Harness bytes={100_000} />));
|
||||
expect(logMock).toHaveBeenLastCalledWith("run-1", 100, 32_000, expect.anything());
|
||||
expect(logMock).toHaveBeenNthCalledWith(3, "run-1", 100, 32_000, expect.anything());
|
||||
expect(truncatedRunIds.has("run-1")).toBe(false);
|
||||
} finally { await act(async () => root.unmount()); }
|
||||
});
|
||||
@@ -198,6 +203,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
const earlier = row(1, "Earlier activity");
|
||||
const recent = row(2, "Latest answer");
|
||||
logMock.mockResolvedValueOnce(result(`${recent}\n`));
|
||||
logMock.mockResolvedValueOnce(result(""));
|
||||
logMock.mockRejectedValueOnce(new Error("Temporary storage failure"));
|
||||
logMock.mockImplementationOnce(async (...args: unknown[]) => {
|
||||
const options = args[3] as { alignStartToRecord?: boolean };
|
||||
@@ -224,7 +230,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
expect.objectContaining({ chunk: "Earlier activity" }),
|
||||
expect.objectContaining({ chunk: "Latest answer" }),
|
||||
]);
|
||||
expect(logMock).toHaveBeenLastCalledWith("run-1", 244_000, 256_000, expect.objectContaining({ alignStartToRecord: true }));
|
||||
expect(logMock).toHaveBeenCalledWith("run-1", 244_000, 256_000, expect.objectContaining({ alignStartToRecord: true }));
|
||||
expect(latest.errorsByRun.size).toBe(0);
|
||||
await act(async () => root.render(<Harness limit={256_000} bytes={520_000} />));
|
||||
expect(logMock).toHaveBeenLastCalledWith("run-1", 500_000, 256_000, expect.not.objectContaining({ alignStartToRecord: true }));
|
||||
@@ -233,40 +239,19 @@ describe("useLiveRunTranscripts", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("waits for a connecting socket to open before closing it during cleanup", async () => {
|
||||
it("registers interests on the shared connection without creating a socket", async () => {
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({
|
||||
companyId: "company-1",
|
||||
runs: [{ id: "run-1", status: "running", adapterType: "codex_local" }],
|
||||
});
|
||||
useLiveRunTranscripts({ companyId: "company-1", runs: [{ id: "run-1", status: "running", adapterType: "codex_local" }] });
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
const socket = FakeWebSocket.instances[0];
|
||||
expect(socket.closeCalls).toHaveLength(0);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
|
||||
expect(socket.closeCalls).toHaveLength(0);
|
||||
|
||||
act(() => {
|
||||
socket.triggerOpen();
|
||||
});
|
||||
|
||||
expect(socket.closeCalls).toEqual([{ code: 1000, reason: "live_run_transcripts_unmount" }]);
|
||||
container.remove();
|
||||
const root = createRoot(document.createElement("div"));
|
||||
await act(async () => root.render(<Harness />));
|
||||
await act(async () => negotiate());
|
||||
expect(sent.at(-1)?.runIds).toEqual(["run-1"]);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
await act(async () => root.unmount());
|
||||
expect(sent.at(-1)?.runIds).toEqual([]);
|
||||
expect(listeners.size).toBe(0);
|
||||
});
|
||||
|
||||
it("treats stored run output as available before transcript chunks finish loading", async () => {
|
||||
@@ -427,7 +412,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
});
|
||||
|
||||
expect(logMock).not.toHaveBeenCalled();
|
||||
expect(FakeWebSocket.instances).toHaveLength(0);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness status="running" />);
|
||||
@@ -436,7 +421,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
|
||||
expect(logMock).toHaveBeenCalledTimes(1);
|
||||
expect(logMock).toHaveBeenCalledWith("run-queued", 0, 256_000, expect.objectContaining({ signal: expect.any(AbortSignal) }));
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
@@ -464,7 +449,7 @@ describe("useLiveRunTranscripts", () => {
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(FakeWebSocket.instances).toHaveLength(0);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
expect(logMock).toHaveBeenCalledWith("run-1", 0, 64_000, expect.objectContaining({ signal: expect.any(AbortSignal) }));
|
||||
|
||||
act(() => {
|
||||
@@ -617,26 +602,13 @@ describe("useLiveRunTranscripts", () => {
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
const socket = FakeWebSocket.instances[0]!;
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
|
||||
const sendLogEvent = (seq: number, chunk: string) => {
|
||||
socket.onmessage?.(
|
||||
new MessageEvent("message", {
|
||||
data: JSON.stringify({
|
||||
companyId: "company-1",
|
||||
type: "heartbeat.run.log",
|
||||
createdAt: "2026-04-20T00:00:01.000Z",
|
||||
payload: {
|
||||
runId: "run-1",
|
||||
ts: "2026-04-20T00:00:01.000Z",
|
||||
stream: "stdout",
|
||||
chunk,
|
||||
seq,
|
||||
},
|
||||
}),
|
||||
}),
|
||||
);
|
||||
sendEvent({
|
||||
companyId: "company-1", type: "heartbeat.run.log", createdAt: "2026-04-20T00:00:01.000Z",
|
||||
payload: { runId: "run-1", ts: "2026-04-20T00:00:01.000Z", stream: "stdout", chunk, seq },
|
||||
});
|
||||
};
|
||||
|
||||
// The websocket races ahead of the poller and its seq-2 chunk arrives
|
||||
@@ -695,26 +667,15 @@ describe("useLiveRunTranscripts", () => {
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
expect(socketConstructed).not.toHaveBeenCalled();
|
||||
expect(buildTranscriptMock).toHaveBeenCalledTimes(2);
|
||||
buildTranscriptMock.mockClear();
|
||||
|
||||
await act(async () => {
|
||||
FakeWebSocket.instances[0]!.onmessage?.(
|
||||
new MessageEvent("message", {
|
||||
data: JSON.stringify({
|
||||
companyId: "company-1",
|
||||
type: "heartbeat.run.log",
|
||||
createdAt: "2026-04-20T00:00:00.000Z",
|
||||
payload: {
|
||||
runId: "run-1",
|
||||
ts: "2026-04-20T00:00:00.000Z",
|
||||
stream: "stdout",
|
||||
chunk: "hello from run 1\n",
|
||||
},
|
||||
}),
|
||||
}),
|
||||
);
|
||||
sendEvent({
|
||||
companyId: "company-1", type: "heartbeat.run.log", createdAt: "2026-04-20T00:00:00.000Z",
|
||||
payload: { runId: "run-1", ts: "2026-04-20T00:00:00.000Z", stream: "stdout", chunk: "hello from run 1\n" },
|
||||
});
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
@@ -839,121 +800,4 @@ describe("useLiveRunTranscripts", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps HTTP log polling when the socket constructor fails and retries later", async () => {
|
||||
vi.useFakeTimers();
|
||||
globalThis.WebSocket = class {
|
||||
constructor() { throw new TypeError("WebSocket is not a constructor"); }
|
||||
} as unknown as typeof WebSocket;
|
||||
const runs = [{ id: "run-1", status: "running", adapterType: "codex_local" }];
|
||||
function Harness() {
|
||||
useLiveRunTranscripts({ companyId: "company-1", runs });
|
||||
return null;
|
||||
}
|
||||
const root = createRoot(document.createElement("div"));
|
||||
try {
|
||||
await act(async () => root.render(<Harness />));
|
||||
expect(logMock).toHaveBeenCalledTimes(1);
|
||||
await act(async () => vi.advanceTimersByTimeAsync(30_000));
|
||||
expect(logMock.mock.calls.length).toBeGreaterThan(1);
|
||||
globalThis.WebSocket = FakeWebSocket as unknown as typeof WebSocket;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(15_000));
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
await act(async () => FakeWebSocket.instances[0].triggerOpen());
|
||||
// Cleanup must not depend on the global constructor remaining available.
|
||||
globalThis.WebSocket = undefined as unknown as typeof WebSocket;
|
||||
} finally {
|
||||
await act(async () => root.unmount());
|
||||
}
|
||||
const calls = logMock.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(30_000));
|
||||
expect(logMock).toHaveBeenCalledTimes(calls);
|
||||
expect(FakeWebSocket.instances[0].closeCalls).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("backs off exponentially when the live event socket keeps failing", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
function Harness({ lastOutputBytes }: { lastOutputBytes?: number }) {
|
||||
useLiveRunTranscripts({
|
||||
companyId: "company-1",
|
||||
runs: [{ id: "run-1", status: "running", adapterType: "codex_local", lastOutputBytes }],
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
const root = createRoot(container);
|
||||
|
||||
await act(async () => {
|
||||
root.render(<Harness />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
|
||||
// Cold backend: every handshake fails. Delays must grow 1.5s → 3s → 6s
|
||||
// instead of hammering a flat interval.
|
||||
await act(async () => {
|
||||
FakeWebSocket.instances[0].triggerClose();
|
||||
await vi.advanceTimersByTimeAsync(1_499);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(1);
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(2);
|
||||
|
||||
await act(async () => {
|
||||
FakeWebSocket.instances[1].triggerClose();
|
||||
await vi.advanceTimersByTimeAsync(2_999);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(2);
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(3);
|
||||
|
||||
await act(async () => {
|
||||
FakeWebSocket.instances[2].triggerClose();
|
||||
await vi.advanceTimersByTimeAsync(5_999);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(3);
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(4);
|
||||
|
||||
// Run-metadata changes restart the socket effect; the progressed delay
|
||||
// must survive the restart instead of resetting to the base delay.
|
||||
await act(async () => {
|
||||
root.render(<Harness lastOutputBytes={512} />);
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(5);
|
||||
await act(async () => {
|
||||
FakeWebSocket.instances[4].triggerClose();
|
||||
await vi.advanceTimersByTimeAsync(11_999);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(5);
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(6);
|
||||
|
||||
// A successful connection resets the backoff to the base delay.
|
||||
await act(async () => {
|
||||
FakeWebSocket.instances[5].triggerOpen();
|
||||
FakeWebSocket.instances[5].triggerClose();
|
||||
await vi.advanceTimersByTimeAsync(1_500);
|
||||
});
|
||||
expect(FakeWebSocket.instances).toHaveLength(7);
|
||||
|
||||
act(() => {
|
||||
root.unmount();
|
||||
});
|
||||
container.remove();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -1,3 +1,5 @@
|
||||
import { TranscriptSequence } from "../../lib/transcript-sequence";
|
||||
import { useTranscriptLiveStream } from "../../context/LiveUpdatesProvider";
|
||||
import { usePageVisibility } from "../../lib/page-visibility";
|
||||
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||
import { readTranscriptRequest } from "./read-transcript-request";
|
||||
@@ -8,8 +10,6 @@ import { instanceSettingsApi } from "../../api/instanceSettings";
|
||||
import { heartbeatsApi } from "../../api/heartbeats";
|
||||
import { buildTranscript, getUIAdapter, onAdapterChange, type RunLogChunk, type TranscriptEntry } from "../../adapters";
|
||||
import { queryKeys } from "../../lib/queryKeys";
|
||||
import { buildSameOriginWebSocketUrl } from "../../lib/websocket-url";
|
||||
import { tryCreateWebSocket } from "../../lib/websocket";
|
||||
import {
|
||||
mergeRunLogChunks,
|
||||
parsePersistedLogContent,
|
||||
@@ -17,17 +17,8 @@ import {
|
||||
type ChunkRetentionBudget,
|
||||
} from "../../lib/run-log-chunks";
|
||||
|
||||
// TODO(perf): this whole hook polls the log/runs endpoints on an interval. The
|
||||
// durable fix is server push (SSE/websocket) for transcript deltas so idle tabs
|
||||
// do no periodic work at all; the constants below only reduce the churn of the
|
||||
// current polling approach.
|
||||
const SOCKET_CONNECTING = 0;
|
||||
const SOCKET_OPEN = 1;
|
||||
const LOG_POLL_INTERVAL_MS = 2000;
|
||||
const LOG_READ_LIMIT_BYTES = 256_000;
|
||||
// When realtime websocket updates are enabled, the frequent log poll is
|
||||
// redundant with the live stream; keep only a slow safety-net poll to cover
|
||||
// gaps and reconnects instead of polling every couple of seconds.
|
||||
const REALTIME_FALLBACK_POLL_INTERVAL_MS = 30_000;
|
||||
const EMPTY_RUN_LOG_CHUNKS: RunLogChunk[] = [];
|
||||
// Retained transcript payload budget for full task views. A byte budget (rather
|
||||
@@ -54,6 +45,7 @@ export interface RunTranscriptSource {
|
||||
hasStoredOutput?: boolean;
|
||||
logBytes?: number | null;
|
||||
lastOutputBytes?: number | null;
|
||||
logRef?: string | null;
|
||||
}
|
||||
|
||||
interface UseLiveRunTranscriptsOptions {
|
||||
@@ -69,6 +61,7 @@ interface UseLiveRunTranscriptsOptions {
|
||||
logPollIntervalMs?: number;
|
||||
logReadLimitBytes?: number;
|
||||
enableRealtimeUpdates?: boolean;
|
||||
enabled?: boolean;
|
||||
}
|
||||
|
||||
function readString(value: unknown): string | null {
|
||||
@@ -104,6 +97,7 @@ export function useLiveRunTranscripts({
|
||||
logPollIntervalMs = LOG_POLL_INTERVAL_MS,
|
||||
logReadLimitBytes = LOG_READ_LIMIT_BYTES,
|
||||
enableRealtimeUpdates = true,
|
||||
enabled = true,
|
||||
}: UseLiveRunTranscriptsOptions) {
|
||||
// Ticker consumers opt into the silent chunk-count cap; full task views use a
|
||||
// byte budget that collapses (not discards) the oldest output when exceeded.
|
||||
@@ -119,9 +113,9 @@ export function useLiveRunTranscripts({
|
||||
() =>
|
||||
runs
|
||||
.map((run) => {
|
||||
const logBytes = typeof run.logBytes === "number" ? run.logBytes : "";
|
||||
const lastOutputBytes = typeof run.lastOutputBytes === "number" ? run.lastOutputBytes : "";
|
||||
return `${run.id}:${run.status}:${run.adapterType}:${run.hasStoredOutput === true ? "1" : "0"}:${logBytes}:${lastOutputBytes}`;
|
||||
const logBytes = isTerminalStatus(run.status) && typeof run.logBytes === "number" ? run.logBytes : "";
|
||||
const lastOutputBytes = isTerminalStatus(run.status) && typeof run.lastOutputBytes === "number" ? run.lastOutputBytes : "";
|
||||
return `${run.id}:${run.status}:${run.adapterType}:${run.hasStoredOutput === true ? "1" : "0"}:${logBytes}:${lastOutputBytes}:${run.logRef ?? ""}`;
|
||||
})
|
||||
.sort((a, b) => a.localeCompare(b))
|
||||
.join(","),
|
||||
@@ -144,6 +138,45 @@ export function useLiveRunTranscripts({
|
||||
const pendingLogRowsByRunRef = useRef(new Map<string, string>());
|
||||
const logOffsetByRunRef = useRef(new Map<string, number>());
|
||||
const logReadLimitByRunRef = useRef(new Map<string, number>());
|
||||
const logIdentityRef = useRef(new Map<string, string>());
|
||||
const retiredLogIdentitiesRef = useRef(new Map<string, Set<string>>());
|
||||
const sequencesRef = useRef(new Map<string, TranscriptSequence>());
|
||||
const sequence = (id: string) => {
|
||||
let value = sequencesRef.current.get(id);
|
||||
if (!value) { value = new TranscriptSequence(); sequencesRef.current.set(id, value); }
|
||||
return value;
|
||||
};
|
||||
const gapPendingRef = useRef(new Set<string>());
|
||||
const refreshByRunRef = useRef(new Map<string, (force?: boolean) => void>());
|
||||
const requestCatchUp = (runId: string) => {
|
||||
if (gapPendingRef.current.has(runId)) return;
|
||||
gapPendingRef.current.add(runId);
|
||||
refreshByRunRef.current.get(runId)?.();
|
||||
};
|
||||
const bindLog = (runId: string, store: unknown, ref: unknown): "same" | "changed" | "stale" => {
|
||||
if (typeof ref !== "string" || !ref) return "same";
|
||||
const identity = `${typeof store === "string" ? store : ""}:${ref}`;
|
||||
const prior = logIdentityRef.current.get(runId);
|
||||
if (prior === identity) return "same";
|
||||
if (retiredLogIdentitiesRef.current.get(runId)?.has(identity)) return "stale";
|
||||
logIdentityRef.current.set(runId, identity);
|
||||
if (!prior) return "same";
|
||||
const retired = retiredLogIdentitiesRef.current.get(runId) ?? new Set<string>();
|
||||
retired.add(prior);
|
||||
if (retired.size > 8) retired.delete(retired.values().next().value!);
|
||||
retiredLogIdentitiesRef.current.set(runId, retired);
|
||||
logOffsetByRunRef.current.set(runId, 0); logReadLimitByRunRef.current.delete(runId);
|
||||
pendingLogRowsByRunRef.current.delete(runId); pendingLogRowsByRunRef.current.delete(`${runId}:records`);
|
||||
trimmedSeqFloorByRunRef.current.delete(runId); sequencesRef.current.delete(runId);
|
||||
for (const key of seenChunkKeysRef.current) if (key.includes(`:${runId}:`)) seenChunkKeysRef.current.delete(key);
|
||||
setChunksByRun(previous => new Map(previous).set(runId, []));
|
||||
setTruncatedRunIds(previous => {
|
||||
if (!previous.has(runId)) return previous;
|
||||
const next = new Set(previous); next.delete(runId); return next;
|
||||
});
|
||||
return "changed";
|
||||
};
|
||||
|
||||
const [truncatedRunIds, setTruncatedRunIds] = useState<ReadonlySet<string>>(new Set());
|
||||
const missingTerminalLogRunIdsRef = useRef(new Set<string>());
|
||||
// PAP-462 B3: buffered runs that dropped out of the `runs` list, mapped to the
|
||||
@@ -152,10 +185,6 @@ export function useLiveRunTranscripts({
|
||||
// the effect again once the nearest deadline elapses so a run that stays gone
|
||||
// is eventually cleaned up even if the `runs` list never changes again.
|
||||
const absenceDeadlineByRunRef = useRef(new Map<string, number>());
|
||||
// Backoff state for the live event socket. Held outside the socket effect
|
||||
// because that effect restarts on run-metadata changes; per-effect state
|
||||
// would reset a progressed delay back to its base mid-outage.
|
||||
const reconnectStateRef = useRef<{ companyId: string; attempt: number } | null>(null);
|
||||
const prevKnownRunIdsRef = useRef(new Set<string>());
|
||||
const [pruneTick, setPruneTick] = useState(0);
|
||||
const transcriptCacheRef = useRef(new Map<string, {
|
||||
@@ -165,6 +194,18 @@ export function useLiveRunTranscripts({
|
||||
parserTick: number;
|
||||
transcript: TranscriptEntry[];
|
||||
}>());
|
||||
const previousCompanyRef = useRef(companyId);
|
||||
useEffect(() => {
|
||||
if (previousCompanyRef.current === companyId) return;
|
||||
previousCompanyRef.current = companyId;
|
||||
setChunksByRun(new Map()); setHydratedRunIds(new Set()); setErrorsByRun(new Map());
|
||||
seenChunkKeysRef.current.clear(); trimmedSeqFloorByRunRef.current.clear();
|
||||
pendingLogRowsByRunRef.current.clear(); logOffsetByRunRef.current.clear(); logReadLimitByRunRef.current.clear();
|
||||
logIdentityRef.current.clear(); retiredLogIdentitiesRef.current.clear(); sequencesRef.current.clear();
|
||||
gapPendingRef.current.clear(); missingTerminalLogRunIdsRef.current.clear();
|
||||
absenceDeadlineByRunRef.current.clear(); prevKnownRunIdsRef.current.clear(); transcriptCacheRef.current.clear();
|
||||
}, [companyId]);
|
||||
|
||||
// Tick counter to force transcript recomputation when dynamic parser loads
|
||||
const [parserTick, setParserTick] = useState(0);
|
||||
useEffect(() => {
|
||||
@@ -294,6 +335,11 @@ export function useLiveRunTranscripts({
|
||||
}
|
||||
}
|
||||
|
||||
for (const map of [logIdentityRef.current, retiredLogIdentitiesRef.current, sequencesRef.current]) {
|
||||
for (const id of map.keys()) if (!retainedRunIds.has(id)) map.delete(id);
|
||||
}
|
||||
for (const id of gapPendingRef.current) if (!retainedRunIds.has(id)) gapPendingRef.current.delete(id);
|
||||
|
||||
// Re-run once the nearest grace window elapses so a run that stays gone is
|
||||
// pruned even if `normalizedRuns` never changes again.
|
||||
if (soonestExpiryMs !== Number.POSITIVE_INFINITY) {
|
||||
@@ -305,193 +351,21 @@ export function useLiveRunTranscripts({
|
||||
}
|
||||
}, [normalizedRuns, pruneTick]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!visible) return;
|
||||
const readableRuns = normalizedRuns.filter(canReadPersistedLog);
|
||||
if (readableRuns.length === 0) return;
|
||||
|
||||
let cancelled = false;
|
||||
const controller = new AbortController();
|
||||
const inFlightRunIds = new Set<string>();
|
||||
|
||||
const readRunLog = async (run: RunTranscriptSource) => {
|
||||
if (missingTerminalLogRunIdsRef.current.has(run.id) || inFlightRunIds.has(run.id)) {
|
||||
return;
|
||||
}
|
||||
inFlightRunIds.add(run.id);
|
||||
let alignStartToRecord = !logOffsetByRunRef.current.has(run.id);
|
||||
let offset = logOffsetByRunRef.current.get(run.id) ?? resolveInitialLogOffset(run, logReadLimitBytes);
|
||||
let rewinding = false;
|
||||
const previousReadLimit = logReadLimitByRunRef.current.get(run.id);
|
||||
if (previousReadLimit !== undefined && logReadLimitBytes > previousReadLimit) {
|
||||
// Expanding a compact tail must reread the earlier range. Retain the
|
||||
// visible chunks; merging by sequence deduplicates the overlapping tail.
|
||||
const expandedOffset = resolveInitialLogOffset(run, logReadLimitBytes);
|
||||
if (expandedOffset < offset) {
|
||||
alignStartToRecord = true;
|
||||
offset = expandedOffset;
|
||||
rewinding = true;
|
||||
}
|
||||
}
|
||||
try {
|
||||
const result = await readTranscriptRequest(
|
||||
(signal) => heartbeatsApi.log(run.id, offset, logReadLimitBytes, {
|
||||
signal,
|
||||
...(alignStartToRecord && offset > 0 ? { alignStartToRecord: true } : {}),
|
||||
}),
|
||||
controller.signal,
|
||||
);
|
||||
if (cancelled) return;
|
||||
|
||||
// Commit the expanded window only after success so retries still align
|
||||
// its first record and failed reads leave the previous parser intact.
|
||||
if (previousReadLimit === undefined || rewinding) {
|
||||
setTruncatedRunIds((previous) => {
|
||||
if (previous.has(run.id) === (offset > 0)) return previous;
|
||||
const next = new Set(previous);
|
||||
if (offset > 0) next.add(run.id);
|
||||
else next.delete(run.id);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
logReadLimitByRunRef.current.set(run.id, logReadLimitBytes);
|
||||
if (rewinding) {
|
||||
pendingLogRowsByRunRef.current.delete(run.id);
|
||||
pendingLogRowsByRunRef.current.delete(`${run.id}:records`);
|
||||
}
|
||||
setErrorsByRun((previous) => {
|
||||
if (!previous.has(run.id)) return previous;
|
||||
const next = new Map(previous);
|
||||
next.delete(run.id);
|
||||
return next;
|
||||
});
|
||||
appendChunks(run.id, parsePersistedLogContent(run.id, result.content, pendingLogRowsByRunRef.current));
|
||||
|
||||
if (result.nextOffset !== undefined) {
|
||||
logOffsetByRunRef.current.set(run.id, result.nextOffset);
|
||||
return;
|
||||
}
|
||||
if (result.content.length > 0) {
|
||||
logOffsetByRunRef.current.set(run.id, offset + result.content.length);
|
||||
}
|
||||
} catch (error) {
|
||||
if (cancelled) return;
|
||||
if (error instanceof ApiError && error.status === 404) {
|
||||
setErrorsByRun((previous) => {
|
||||
if (!previous.has(run.id)) return previous;
|
||||
const next = new Map(previous);
|
||||
next.delete(run.id);
|
||||
return next;
|
||||
});
|
||||
// A newly started run may not have created its log yet.
|
||||
if (isTerminalStatus(run.status)) missingTerminalLogRunIdsRef.current.add(run.id);
|
||||
} else {
|
||||
setErrorsByRun((previous) => {
|
||||
if (previous.has(run.id)) return previous;
|
||||
const next = new Map(previous);
|
||||
next.set(run.id, error instanceof Error ? error : new Error("Run history could not be loaded"));
|
||||
return next;
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
inFlightRunIds.delete(run.id);
|
||||
if (!cancelled) {
|
||||
setHydratedRunIds((prev) => {
|
||||
if (prev.has(run.id)) return prev;
|
||||
const next = new Set(prev);
|
||||
next.add(run.id);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const readAll = async () => {
|
||||
await Promise.all(readableRuns.map((run) => readRunLog(run)));
|
||||
};
|
||||
|
||||
void readAll();
|
||||
const activeRuns = readableRuns.filter((run) => run.status === "running");
|
||||
// The realtime websocket is the primary live source when enabled, so the
|
||||
// recurring poll only needs to run as a slow fallback rather than doubling
|
||||
// the live update work every couple of seconds.
|
||||
const effectivePollMs = enableRealtimeUpdates
|
||||
? Math.max(logPollIntervalMs, REALTIME_FALLBACK_POLL_INTERVAL_MS)
|
||||
: logPollIntervalMs;
|
||||
const interval = activeRuns.length > 0 && effectivePollMs > 0
|
||||
? window.setInterval(() => {
|
||||
void Promise.all(activeRuns.map((run) => readRunLog(run)));
|
||||
}, effectivePollMs)
|
||||
: null;
|
||||
|
||||
return () => {
|
||||
cancelled = true;
|
||||
controller.abort();
|
||||
if (interval !== null) window.clearInterval(interval);
|
||||
};
|
||||
}, [visible, enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey, retryGeneration]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!visible || !enableRealtimeUpdates) return;
|
||||
if (!companyId || activeRunIds.size === 0) return;
|
||||
|
||||
let closed = false;
|
||||
let reconnectTimer: number | null = null;
|
||||
let socket: WebSocket | null = null;
|
||||
|
||||
// The attempt counter lives in a ref keyed to the company: this effect
|
||||
// restarts whenever run metadata changes, and a per-effect counter would
|
||||
// reset the backoff to its base delay mid-outage on every such restart.
|
||||
if (reconnectStateRef.current?.companyId !== companyId) {
|
||||
reconnectStateRef.current = { companyId, attempt: 0 };
|
||||
}
|
||||
const reconnectState = reconnectStateRef.current;
|
||||
|
||||
// Exponential backoff (1.5s → 15s cap), mirroring LiveUpdatesProvider.
|
||||
// A flat retry hammers a backend that is still cold-starting — every
|
||||
// failed handshake immediately queues the next one, so a stack that
|
||||
// takes a minute to come up sees a steady stream of doomed connections.
|
||||
const scheduleReconnect = () => {
|
||||
if (closed) return;
|
||||
reconnectState.attempt += 1;
|
||||
const delayMs = Math.min(15_000, 1_500 * 2 ** Math.min(reconnectState.attempt - 1, 4));
|
||||
reconnectTimer = window.setTimeout(connect, delayMs);
|
||||
};
|
||||
|
||||
const connect = () => {
|
||||
if (closed) return;
|
||||
const url = buildSameOriginWebSocketUrl(
|
||||
`/api/companies/${encodeURIComponent(companyId)}/events/ws`,
|
||||
);
|
||||
socket = tryCreateWebSocket(url);
|
||||
if (!socket) {
|
||||
scheduleReconnect();
|
||||
return;
|
||||
}
|
||||
|
||||
socket.onopen = () => {
|
||||
if (closed) return;
|
||||
reconnectState.attempt = 0;
|
||||
};
|
||||
|
||||
socket.onmessage = (message) => {
|
||||
const raw = typeof message.data === "string" ? message.data : "";
|
||||
if (!raw) return;
|
||||
|
||||
let event: LiveEvent;
|
||||
try {
|
||||
event = JSON.parse(raw) as LiveEvent;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
const stream = useTranscriptLiveStream([...activeRunIds], { companyId, enabled: enabled && enableRealtimeUpdates }, event => {
|
||||
if (event.companyId !== companyId) return;
|
||||
const payload = event.payload ?? {};
|
||||
const runId = readString(payload["runId"]);
|
||||
if (!runId || !activeRunIds.has(runId)) return;
|
||||
if (!runId) return;
|
||||
if (!runById.has(runId)) return;
|
||||
|
||||
if (event.type === "heartbeat.run.log") {
|
||||
const binding = bindLog(runId, payload.logStore, payload.logRef);
|
||||
if (binding === "stale") return;
|
||||
const seq = readChunkSeq(payload.seq);
|
||||
if (payload.truncated === true) sequence(runId).requireHistory(seq);
|
||||
const gap = sequence(runId).observe(seq);
|
||||
if (binding === "changed" || payload.truncated === true || gap) requestCatchUp(runId);
|
||||
if (payload.truncated === true) return;
|
||||
const chunk = readString(payload["chunk"]);
|
||||
if (!chunk) return;
|
||||
const ts = readString(payload["ts"]) ?? event.createdAt;
|
||||
@@ -526,6 +400,7 @@ export function useLiveRunTranscripts({
|
||||
|
||||
if (event.type === "heartbeat.run.status") {
|
||||
const status = readString(payload["status"]) ?? "updated";
|
||||
if (isTerminalStatus(status)) refreshByRunRef.current.get(runId)?.(true);
|
||||
appendChunks(runId, [{
|
||||
ts: event.createdAt,
|
||||
stream: isTerminalStatus(status) && status !== "succeeded" ? "stderr" : "system",
|
||||
@@ -533,39 +408,154 @@ export function useLiveRunTranscripts({
|
||||
dedupeKey: `socket:status:${runId}:${status}:${readString(payload["finishedAt"]) ?? ""}`,
|
||||
}]);
|
||||
}
|
||||
};
|
||||
});
|
||||
|
||||
socket.onerror = () => {
|
||||
socket?.close();
|
||||
};
|
||||
useEffect(() => {
|
||||
if (!visible || !enabled) return;
|
||||
const readableRuns = normalizedRuns.filter(canReadPersistedLog);
|
||||
if (readableRuns.length === 0) return;
|
||||
|
||||
socket.onclose = () => {
|
||||
scheduleReconnect();
|
||||
};
|
||||
};
|
||||
let cancelled = false;
|
||||
const controller = new AbortController();
|
||||
const inFlightRunIds = new Set<string>();
|
||||
const retryTimers = new Map<string, number>();
|
||||
const pendingRefresh = new Set<string>();
|
||||
|
||||
connect();
|
||||
|
||||
return () => {
|
||||
closed = true;
|
||||
if (reconnectTimer !== null) window.clearTimeout(reconnectTimer);
|
||||
if (socket) {
|
||||
socket.onmessage = null;
|
||||
socket.onerror = null;
|
||||
socket.onclose = null;
|
||||
if (socket.readyState === SOCKET_CONNECTING) {
|
||||
// Defer the close until the handshake completes so the browser
|
||||
// does not emit a noisy "closed before the connection is established"
|
||||
// warning during rapid run teardown.
|
||||
socket.onopen = () => {
|
||||
socket?.close(1000, "live_run_transcripts_unmount");
|
||||
};
|
||||
} else if (socket.readyState === SOCKET_OPEN) {
|
||||
socket.close(1000, "live_run_transcripts_unmount");
|
||||
const readRunLog = async (run: RunTranscriptSource) => {
|
||||
if (cancelled || missingTerminalLogRunIdsRef.current.has(run.id) || inFlightRunIds.has(run.id)) {
|
||||
return;
|
||||
}
|
||||
pendingRefresh.delete(run.id);
|
||||
window.clearTimeout(retryTimers.get(run.id));
|
||||
retryTimers.delete(run.id);
|
||||
inFlightRunIds.add(run.id);
|
||||
let failed = false;
|
||||
let alignStartToRecord = !logOffsetByRunRef.current.has(run.id);
|
||||
let offset = logOffsetByRunRef.current.get(run.id) ?? resolveInitialLogOffset(run, logReadLimitBytes);
|
||||
let rewinding = false;
|
||||
const previousReadLimit = logReadLimitByRunRef.current.get(run.id);
|
||||
if (previousReadLimit !== undefined && logReadLimitBytes > previousReadLimit) {
|
||||
const expandedOffset = resolveInitialLogOffset(run, logReadLimitBytes);
|
||||
if (expandedOffset < offset) {
|
||||
alignStartToRecord = true;
|
||||
offset = expandedOffset;
|
||||
rewinding = true;
|
||||
}
|
||||
}
|
||||
let recordInitialWindow = previousReadLimit === undefined || rewinding;
|
||||
try {
|
||||
for (;;) {
|
||||
const result = await readTranscriptRequest(
|
||||
(signal) => heartbeatsApi.log(run.id, offset, logReadLimitBytes, {
|
||||
signal,
|
||||
...(alignStartToRecord && offset > 0 ? { alignStartToRecord: true } : {}),
|
||||
}), controller.signal,
|
||||
);
|
||||
if (cancelled) return;
|
||||
const binding = bindLog(run.id, result.store, result.logRef);
|
||||
if (binding === "stale") { failed = true; return; }
|
||||
if (binding === "changed" && offset !== 0) { offset = 0; continue; }
|
||||
if (recordInitialWindow || binding === "changed") {
|
||||
const windowIsTruncated = offset > 0;
|
||||
setTruncatedRunIds(previous => {
|
||||
if (previous.has(run.id) === windowIsTruncated) return previous;
|
||||
const next = new Set(previous);
|
||||
if (windowIsTruncated) next.add(run.id);
|
||||
else next.delete(run.id);
|
||||
return next;
|
||||
});
|
||||
recordInitialWindow = false;
|
||||
}
|
||||
logReadLimitByRunRef.current.set(run.id, logReadLimitBytes);
|
||||
if (rewinding) {
|
||||
pendingLogRowsByRunRef.current.delete(run.id);
|
||||
pendingLogRowsByRunRef.current.delete(`${run.id}:records`);
|
||||
rewinding = false;
|
||||
}
|
||||
alignStartToRecord = false;
|
||||
setErrorsByRun(previous => {
|
||||
if (!previous.has(run.id)) return previous;
|
||||
const next = new Map(previous); next.delete(run.id); return next;
|
||||
});
|
||||
const rows = parsePersistedLogContent(run.id, result.content, pendingLogRowsByRunRef.current);
|
||||
appendChunks(run.id, rows);
|
||||
for (const row of rows) sequence(run.id).confirm(row.seq);
|
||||
const next = result.nextOffset ?? offset + new TextEncoder().encode(result.content).byteLength;
|
||||
logOffsetByRunRef.current.set(run.id, next);
|
||||
if (result.nextOffset === undefined || result.content.length === 0 || next <= offset) break;
|
||||
offset = next;
|
||||
}
|
||||
failed = sequence(run.id).needsCatchup;
|
||||
if (!failed) gapPendingRef.current.delete(run.id);
|
||||
} catch (error) {
|
||||
if (cancelled) return;
|
||||
failed = true;
|
||||
if (error instanceof ApiError && error.status === 404) {
|
||||
setErrorsByRun((previous) => {
|
||||
if (!previous.has(run.id)) return previous;
|
||||
const next = new Map(previous);
|
||||
next.delete(run.id);
|
||||
return next;
|
||||
});
|
||||
// A newly started run may not have created its log yet.
|
||||
if (isTerminalStatus(run.status)) { missingTerminalLogRunIdsRef.current.add(run.id); failed = false; }
|
||||
} else {
|
||||
setErrorsByRun((previous) => {
|
||||
if (previous.has(run.id)) return previous;
|
||||
const next = new Map(previous);
|
||||
next.set(run.id, error instanceof Error ? error : new Error("Run history could not be loaded"));
|
||||
return next;
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
inFlightRunIds.delete(run.id);
|
||||
if (!cancelled) {
|
||||
if (pendingRefresh.has(run.id)) {
|
||||
missingTerminalLogRunIdsRef.current.delete(run.id);
|
||||
void readRunLog(run);
|
||||
} else if (failed && stream.healthy) {
|
||||
const timer = window.setTimeout(() => { retryTimers.delete(run.id); void readRunLog(run); }, LOG_POLL_INTERVAL_MS);
|
||||
retryTimers.set(run.id, timer);
|
||||
}
|
||||
setHydratedRunIds((prev) => {
|
||||
if (prev.has(run.id)) return prev;
|
||||
const next = new Set(prev);
|
||||
next.add(run.id);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
}, [visible, activeRunIds, companyId, enableRealtimeUpdates, runById]);
|
||||
|
||||
for (const run of readableRuns) {
|
||||
refreshByRunRef.current.set(run.id, (force = false) => {
|
||||
if (force) { pendingRefresh.add(run.id); missingTerminalLogRunIdsRef.current.delete(run.id); }
|
||||
void readRunLog(run);
|
||||
});
|
||||
void readRunLog(run);
|
||||
}
|
||||
const activeRuns = readableRuns.filter((run) => run.status === "running");
|
||||
// The realtime websocket is the primary live source when enabled, so the
|
||||
// recurring poll only needs to run as a slow fallback rather than doubling
|
||||
// the live update work every couple of seconds.
|
||||
const effectivePollMs = enableRealtimeUpdates
|
||||
? Math.max(logPollIntervalMs, REALTIME_FALLBACK_POLL_INTERVAL_MS)
|
||||
: logPollIntervalMs;
|
||||
const interval = !stream.healthy && activeRuns.length > 0 && effectivePollMs > 0
|
||||
? window.setInterval(() => {
|
||||
void Promise.all(activeRuns.map((run) => readRunLog(run)));
|
||||
}, effectivePollMs)
|
||||
: null;
|
||||
|
||||
return () => {
|
||||
cancelled = true;
|
||||
refreshByRunRef.current.clear();
|
||||
controller.abort();
|
||||
for (const timer of retryTimers.values()) window.clearTimeout(timer);
|
||||
if (interval !== null) window.clearInterval(interval);
|
||||
};
|
||||
}, [visible, enableRealtimeUpdates, logPollIntervalMs, logReadLimitBytes, normalizedRuns, runIdsKey, retryGeneration, enabled, companyId, stream.healthy, stream.epoch]);
|
||||
|
||||
|
||||
const transcriptByRun = useMemo(() => {
|
||||
const next = new Map<string, TranscriptEntry[]>();
|
||||
|
||||
@@ -0,0 +1,290 @@
|
||||
// @vitest-environment jsdom
|
||||
import { act } from "react";
|
||||
import { createRoot, type Root } from "react-dom/client";
|
||||
import { afterEach, beforeEach, expect, it, vi } from "vitest";
|
||||
import type { HeartbeatRunEvent, LiveEvent } from "@paperclipai/shared";
|
||||
import { __liveUpdatesTestUtils, type CompanyLiveEventHandler } from "../../context/LiveUpdatesProvider";
|
||||
import { LiveTranscriptInterests } from "../../lib/live-transcript-interests";
|
||||
import { useNativeRunTranscripts } from "./useNativeRunTranscripts";
|
||||
import { nativeRunEventsToTranscript } from "./native-run-events";
|
||||
|
||||
const events = vi.hoisted(() => vi.fn());
|
||||
const conversationEvents = vi.hoisted(() => vi.fn());
|
||||
vi.mock("../../api/heartbeats", () => ({ heartbeatsApi: { events, conversationEvents } }));
|
||||
(globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true;
|
||||
const { LiveEventSubscriptionContext } = __liveUpdatesTestUtils;
|
||||
let root: Root;
|
||||
let store: LiveTranscriptInterests;
|
||||
let listeners: Set<CompanyLiveEventHandler>;
|
||||
let sent: Array<{ revision?: number; runIds?: string[] }>;
|
||||
let latest: ReturnType<typeof useNativeRunTranscripts>;
|
||||
let subscription: { companyId: string; transcripts: LiveTranscriptInterests; subscribe: (fn: CompanyLiveEventHandler) => () => void };
|
||||
const row = (seq: number, message = `row ${seq}`, runId = "run-1"): HeartbeatRunEvent => ({
|
||||
id: seq, seq, runId, companyId: "company-1", agentId: "agent-1", eventType: "item.completed",
|
||||
stream: null, level: null, color: null, message, payload: { prpEvent: {
|
||||
schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceEventId: `event-${seq}`,
|
||||
sourceSeq: seq, sourceInstanceId: "runner-1", sourceKind: "runner", runId,
|
||||
normalizedSessionId: "session-1", eventType: "item.completed", itemId: `message-${seq}`,
|
||||
emittedAt: "2026-10-24T00:00:00Z", payload: { kind: "agentMessage", channel: "progress", text: message },
|
||||
} }, createdAt: new Date("2026-10-24T00:00:00Z"),
|
||||
});
|
||||
function emit(seq: number, signalOnly = false, kind = "agentMessage") {
|
||||
const value = row(seq);
|
||||
const envelope = value.payload!.prpEvent as { payload: Record<string, unknown> };
|
||||
envelope.payload = { ...envelope.payload, kind };
|
||||
const event: LiveEvent = { id: seq, companyId: value.companyId, type: "heartbeat.run.event", createdAt: value.createdAt.toISOString(), payload: signalOnly
|
||||
? { runId: value.runId, agentId: value.agentId, eventType: value.eventType, seq }
|
||||
: { ...value } };
|
||||
for (const listener of listeners) listener(event);
|
||||
}
|
||||
function Probe({ companyId = "company-1", enabled = true, older = false, compact = false }: { companyId?: string; enabled?: boolean; older?: boolean; compact?: boolean }) {
|
||||
latest = useNativeRunTranscripts([
|
||||
{ id: "run-1", status: "running", runtimeMode: "native", conversationOnly: compact },
|
||||
...(older ? [{ id: "older", status: "succeeded", runtimeMode: "native" as const }] : []),
|
||||
], { companyId, enabled });
|
||||
return null;
|
||||
}
|
||||
async function render(companyId = "company-1", enabled = true, older = false, compact = false) {
|
||||
await act(async () => root.render(<LiveEventSubscriptionContext.Provider value={subscription}><Probe companyId={companyId} enabled={enabled} older={older} compact={compact} /></LiveEventSubscriptionContext.Provider>));
|
||||
}
|
||||
async function negotiate() {
|
||||
await act(async () => {
|
||||
store.open({ send: text => sent.push(JSON.parse(text)) });
|
||||
store.receive({ type: "live.capabilities", version: 1, capabilities: ["run-interests"] });
|
||||
store.receive({ type: "live.subscribed", version: 1, revision: sent.at(-1)?.revision });
|
||||
});
|
||||
}
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
vi.spyOn(document, "visibilityState", "get").mockReturnValue("visible");
|
||||
events.mockReset().mockResolvedValue([]);
|
||||
conversationEvents.mockReset().mockResolvedValue([]);
|
||||
root = createRoot(document.createElement("div"));
|
||||
store = new LiveTranscriptInterests(); listeners = new Set(); sent = [];
|
||||
subscription = { companyId: "company-1", transcripts: store, subscribe(fn) { listeners.add(fn); return () => { listeners.delete(fn); }; } };
|
||||
});
|
||||
afterEach(async () => { await act(async () => root.unmount()); vi.restoreAllMocks(); vi.useRealTimers(); });
|
||||
|
||||
it("merges history arriving after live output without losing rows or duplicating a sequence", async () => {
|
||||
let complete!: (rows: HeartbeatRunEvent[]) => void;
|
||||
events.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; }));
|
||||
await render();
|
||||
await act(async () => emit(3));
|
||||
expect(events).toHaveBeenCalledTimes(1);
|
||||
await act(async () => complete([row(1), row(2), row(3, "canonical row 3")]));
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1), row(2), row(3, "canonical row 3")]));
|
||||
});
|
||||
|
||||
it.each(["agentMessage", "reasoning"])("streams %s deltas after hydration without recurring history reads", async kind => {
|
||||
events.mockResolvedValueOnce([row(1)]);
|
||||
await render(); await negotiate();
|
||||
const reads = events.mock.calls.length;
|
||||
const delta = row(2, "still writing");
|
||||
delta.eventType = "item.delta";
|
||||
const envelope = delta.payload!.prpEvent as Record<string, unknown>;
|
||||
envelope.eventType = "item.delta";
|
||||
envelope.payload = { kind, text: "still writing" };
|
||||
await act(async () => {
|
||||
for (const listener of listeners) listener({ id: 2, companyId: delta.companyId,
|
||||
type: "heartbeat.run.event", createdAt: delta.createdAt.toISOString(), payload: { ...delta } });
|
||||
await vi.advanceTimersByTimeAsync(90_000);
|
||||
});
|
||||
expect(JSON.stringify(latest.transcriptByRun.get("run-1"))).toContain("still writing");
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
});
|
||||
|
||||
it("fetches a consecutive native commit signal and never replaces its complete row with a delayed signal", async () => {
|
||||
events.mockResolvedValueOnce([row(1)]);
|
||||
await render(); await negotiate();
|
||||
const reads = events.mock.calls.length;
|
||||
events.mockResolvedValueOnce([row(2, "completed output")]);
|
||||
await act(async () => emit(2, true));
|
||||
expect(events).toHaveBeenCalledTimes(reads + 1);
|
||||
expect(events.mock.calls.at(-1)?.slice(0, 3)).toEqual(["run-1", 1, 1_000]);
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1), row(2, "completed output")]));
|
||||
await act(async () => { emit(2, true); emit(1, true); await vi.advanceTimersByTimeAsync(90_000); });
|
||||
expect(events).toHaveBeenCalledTimes(reads + 1);
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1), row(2, "completed output")]));
|
||||
});
|
||||
|
||||
it("retains a signal arriving during an older catch-up until its canonical row is visible", async () => {
|
||||
await render(); await negotiate();
|
||||
let complete!: (rows: HeartbeatRunEvent[]) => void;
|
||||
events.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; }));
|
||||
await act(async () => emit(1, true));
|
||||
const reads = events.mock.calls.length;
|
||||
await act(async () => emit(2, true));
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
await act(async () => complete([row(1)]));
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1)]));
|
||||
events.mockRejectedValueOnce(new Error("history temporarily unavailable"));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(2_000));
|
||||
expect(latest.errorsByRun.has("run-1")).toBe(true);
|
||||
events.mockResolvedValueOnce([row(2)]);
|
||||
await act(async () => vi.advanceTimersByTimeAsync(2_000));
|
||||
expect(latest.errorsByRun.size).toBe(0);
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1), row(2)]));
|
||||
const recoveredReads = events.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(events).toHaveBeenCalledTimes(recoveredReads);
|
||||
});
|
||||
|
||||
it("renders complete live PRP rows without initiating or scheduling history reads", async () => {
|
||||
await render(); await negotiate();
|
||||
const reads = events.mock.calls.length;
|
||||
await act(async () => { emit(1); emit(2); await vi.advanceTimersByTimeAsync(90_000); });
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1), row(2)]));
|
||||
});
|
||||
|
||||
function emitTerminal() {
|
||||
const event: LiveEvent = { id: 100, companyId: "company-1", type: "heartbeat.run.status",
|
||||
createdAt: new Date().toISOString(), payload: { runId: "run-1", status: "succeeded" } };
|
||||
for (const listener of listeners) listener(event);
|
||||
}
|
||||
|
||||
it("finishes slow historical pages while a sibling receives repeated commit and terminal notifications", async () => {
|
||||
await render(); await negotiate();
|
||||
const olderReads: Array<{ cursor: number; signal: AbortSignal; resolve: (rows: HeartbeatRunEvent[]) => void }> = [];
|
||||
events.mockImplementation((runId: string, cursor: number, _limit: number, { signal }: { signal: AbortSignal }) => {
|
||||
if (runId === "older") return new Promise(resolve => olderReads.push({ cursor, signal, resolve }));
|
||||
return Promise.resolve([row(cursor + 1)]);
|
||||
});
|
||||
await render("company-1", true, true);
|
||||
expect(olderReads).toHaveLength(1);
|
||||
await act(async () => emit(2, true));
|
||||
expect(olderReads[0].signal.aborted).toBe(false);
|
||||
expect(olderReads).toHaveLength(1);
|
||||
const firstPage = Array.from({ length: 1_000 }, (_, i) => row(i + 1, `historical ${i + 1}`, "older"));
|
||||
await act(async () => olderReads[0].resolve(firstPage));
|
||||
expect(olderReads).toHaveLength(2);
|
||||
expect(olderReads[1].cursor).toBe(1_000);
|
||||
expect(latest.hydratedRunIds.has("older")).toBe(false);
|
||||
await act(async () => emit(3, true));
|
||||
await act(async () => emit(4, true));
|
||||
await act(async () => emitTerminal());
|
||||
expect(olderReads[1].signal.aborted).toBe(false);
|
||||
expect(olderReads).toHaveLength(2);
|
||||
const lastRow = row(1_001, "last historical output", "older");
|
||||
await act(async () => olderReads[1].resolve([lastRow]));
|
||||
expect(latest.hydratedRunIds.has("older")).toBe(true);
|
||||
expect(latest.transcriptByRun.get("older")).toEqual(nativeRunEventsToTranscript([...firstPage, lastRow]));
|
||||
const reads = events.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
});
|
||||
|
||||
it("queues one terminal catch-up behind an in-flight read without aborting or overlapping it", async () => {
|
||||
await render(); await negotiate();
|
||||
let complete!: (rows: HeartbeatRunEvent[]) => void;
|
||||
let activeSignal!: AbortSignal;
|
||||
events.mockImplementationOnce((_runId, _cursor, _limit, { signal }) => {
|
||||
activeSignal = signal;
|
||||
return new Promise(resolve => { complete = resolve; });
|
||||
});
|
||||
await act(async () => emit(1, true));
|
||||
const reads = events.mock.calls.length;
|
||||
await act(async () => { emitTerminal(); emitTerminal(); });
|
||||
expect(activeSignal.aborted).toBe(false);
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
events.mockResolvedValueOnce([row(2, "terminal output")]);
|
||||
await act(async () => complete([row(1)]));
|
||||
expect(events).toHaveBeenCalledTimes(reads + 1);
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1), row(2, "terminal output")]));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(events).toHaveBeenCalledTimes(reads + 1);
|
||||
});
|
||||
|
||||
it("performs no recurring history reads after an acknowledged catch-up, and retries a failed catch-up", async () => {
|
||||
await render(); await negotiate();
|
||||
const reads = events.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
events.mockRejectedValueOnce(new Error("temporary history outage"));
|
||||
await act(async () => {
|
||||
store.receive({ type: "live.resync", version: 1, revision: sent.at(-1)?.revision });
|
||||
store.receive({ type: "live.subscribed", version: 1, revision: sent.at(-1)?.revision });
|
||||
});
|
||||
expect(latest.errorsByRun.has("run-1")).toBe(true);
|
||||
await act(async () => vi.advanceTimersByTimeAsync(2_000));
|
||||
expect(latest.errorsByRun.size).toBe(0);
|
||||
const recoveredReads = events.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(events).toHaveBeenCalledTimes(recoveredReads);
|
||||
});
|
||||
|
||||
it("withdraws disabled views and rejects events from the previous company", async () => {
|
||||
await render(); await negotiate();
|
||||
await act(async () => emit(1));
|
||||
expect(latest.transcriptByRun.get("run-1")).toHaveLength(1);
|
||||
await render("company-1", false);
|
||||
expect(sent.at(-1)?.runIds).toEqual([]);
|
||||
const reads = events.mock.calls.length;
|
||||
await act(async () => { emit(2); await vi.advanceTimersByTimeAsync(30_000); });
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
expect(latest.transcriptByRun.get("run-1")).toHaveLength(1);
|
||||
await render("company-2");
|
||||
await act(async () => emit(3));
|
||||
expect(latest.transcriptByRun.get("run-1") ?? []).toEqual([]);
|
||||
});
|
||||
|
||||
|
||||
it("finishes catch-up while unrelated consumers repeatedly replace shared interests", async () => {
|
||||
await render(); await negotiate();
|
||||
let complete!: (rows: HeartbeatRunEvent[]) => void;
|
||||
events.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; }));
|
||||
await act(async () => emit(1, true));
|
||||
const reads = events.mock.calls.length;
|
||||
const signal = events.mock.calls.at(-1)![3].signal as AbortSignal;
|
||||
for (let index = 0; index < 4; index++) {
|
||||
let release!: () => void;
|
||||
await act(async () => { release = store.acquire([`unrelated-${index}`]); });
|
||||
expect(signal.aborted).toBe(false);
|
||||
await act(async () => store.receive({ type: "live.subscribed", version: 1, revision: sent.at(-1)?.revision }));
|
||||
await act(async () => { release(); });
|
||||
await act(async () => store.receive({ type: "live.subscribed", version: 1, revision: sent.at(-1)?.revision }));
|
||||
expect(signal.aborted).toBe(false);
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
}
|
||||
await act(async () => complete([row(1, "uninterrupted catch-up")]));
|
||||
expect(latest.transcriptByRun.get("run-1")).toEqual(nativeRunEventsToTranscript([row(1, "uninterrupted catch-up")]));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(events).toHaveBeenCalledTimes(reads);
|
||||
});
|
||||
|
||||
it.each([false, true])("settles compact recovery through filtered-out activity without recurring reads (signal only: %s)", async signalOnly => {
|
||||
conversationEvents.mockResolvedValueOnce([row(2, "Saved answer")]);
|
||||
await render("company-1", true, false, true);
|
||||
await negotiate();
|
||||
await act(async () => store.close());
|
||||
conversationEvents.mockImplementation((_runId, after: number) => Promise.resolve(after < 6 ? [row(6, "Recovered answer")] : []));
|
||||
await negotiate();
|
||||
await act(async () => emit(10, signalOnly, "reasoning"));
|
||||
const settledReads = conversationEvents.mock.calls.length;
|
||||
expect(latest.transcriptByRun.get("run-1")?.filter(entry => entry.kind === "assistant").map(entry => entry.text))
|
||||
.toEqual(["Saved answer", "Recovered answer"]);
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(conversationEvents).toHaveBeenCalledTimes(settledReads);
|
||||
expect(events).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not skip compact conversation rows committed during an older catch-up", async () => {
|
||||
conversationEvents.mockResolvedValueOnce([row(2, "Saved answer")]);
|
||||
await render("company-1", true, false, true);
|
||||
await negotiate();
|
||||
let complete!: (rows: HeartbeatRunEvent[]) => void;
|
||||
conversationEvents.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; }));
|
||||
await act(async () => emit(4, true, "reasoning"));
|
||||
const reads = conversationEvents.mock.calls.length;
|
||||
await act(async () => emit(6, true, "reasoning"));
|
||||
expect(conversationEvents).toHaveBeenCalledTimes(reads);
|
||||
conversationEvents.mockImplementation((_runId, after: number) => Promise.resolve(after < 5 ? [row(5, "Answer committed during catch-up")] : []));
|
||||
await act(async () => complete([]));
|
||||
await act(async () => vi.advanceTimersByTimeAsync(2_000));
|
||||
expect(conversationEvents.mock.calls.at(-1)?.slice(0, 3)).toEqual(["run-1", 4, 1_000]);
|
||||
expect(latest.transcriptByRun.get("run-1")?.filter(entry => entry.kind === "assistant").map(entry => entry.text))
|
||||
.toEqual(["Saved answer", "Answer committed during catch-up"]);
|
||||
const settledReads = conversationEvents.mock.calls.length;
|
||||
await act(async () => vi.advanceTimersByTimeAsync(90_000));
|
||||
expect(conversationEvents).toHaveBeenCalledTimes(settledReads);
|
||||
});
|
||||
@@ -1,3 +1,7 @@
|
||||
import { TranscriptSequence } from "../../lib/transcript-sequence";
|
||||
import { useTranscriptLiveStream } from "../../context/LiveUpdatesProvider";
|
||||
import { usePageVisibility } from "../../lib/page-visibility";
|
||||
import { liveRunEvent, mergeRunEvents } from "../../lib/live-run-event";
|
||||
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||
import type { HeartbeatRunEvent } from "@paperclipai/shared";
|
||||
import type { TranscriptEntry } from "@/adapters";
|
||||
@@ -24,7 +28,8 @@ function isLive(status: string): boolean {
|
||||
return status === "queued" || status === "running";
|
||||
}
|
||||
|
||||
export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource[]) {
|
||||
export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource[], options: { companyId?: string | null; enabled?: boolean } = {}) {
|
||||
const { visible } = usePageVisibility();
|
||||
const nativeRunsKey = runs
|
||||
.filter((run) => run.runtimeMode === "native")
|
||||
.map((run) => `${run.id}:${run.status}:${Boolean(run.conversationOnly)}`)
|
||||
@@ -46,15 +51,54 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource
|
||||
const cursorByRunRef = useRef(new Map<string, number>());
|
||||
const conversationOnlyByRunRef = useRef(new Map<string, boolean>());
|
||||
|
||||
const sequencesRef = useRef(new Map<string, TranscriptSequence>());
|
||||
const sequence = (id: string) => {
|
||||
let value = sequencesRef.current.get(id);
|
||||
if (!value) { value = new TranscriptSequence(); sequencesRef.current.set(id, value); }
|
||||
return value;
|
||||
};
|
||||
const gapPendingRef = useRef(new Set<string>());
|
||||
const historyReadsRef = useRef(new Map<string, AbortSignal>());
|
||||
const refreshByRunRef = useRef(new Map<string, (force?: boolean) => void>());
|
||||
const previousCompanyRef = useRef(options.companyId);
|
||||
useEffect(() => {
|
||||
if (previousCompanyRef.current === options.companyId) return;
|
||||
previousCompanyRef.current = options.companyId;
|
||||
setEventsByRun(new Map()); setErrorsByRun(new Map()); setHydratedRunIds(new Set()); setHydratedStatusByRun(new Map());
|
||||
projectionCacheRef.current.clear(); cursorByRunRef.current.clear(); conversationOnlyByRunRef.current.clear();
|
||||
sequencesRef.current.clear(); gapPendingRef.current.clear();
|
||||
}, [options.companyId]);
|
||||
const stream = useTranscriptLiveStream(nativeRuns.filter(run => isLive(run.status)).map(run => run.id), options, event => {
|
||||
const runId = typeof event.payload.runId === "string" ? event.payload.runId : null;
|
||||
if (!runId || !nativeRuns.some(run => run.id === runId)) return;
|
||||
if (event.type === "heartbeat.run.status" && typeof event.payload.status === "string" && !isLive(event.payload.status)) {
|
||||
refreshByRunRef.current.get(runId)?.(true); return;
|
||||
}
|
||||
const row = liveRunEvent(event);
|
||||
if (!row) return;
|
||||
const cursor = sequence(runId);
|
||||
const envelope = row.payload?.prpEvent;
|
||||
const hasTranscriptBody = envelope !== null && typeof envelope === "object" && !Array.isArray(envelope);
|
||||
if (!hasTranscriptBody) cursor.requireHistory(row.seq);
|
||||
if (cursor.observe(row.seq) && !gapPendingRef.current.has(runId)) {
|
||||
gapPendingRef.current.add(runId);
|
||||
refreshByRunRef.current.get(runId)?.();
|
||||
}
|
||||
if (hasTranscriptBody) setEventsByRun(previous => new Map(previous).set(runId, mergeRunEvents(previous.get(runId) ?? [], [row])));
|
||||
});
|
||||
|
||||
|
||||
useEffect(() => {
|
||||
if (!visible || options.enabled === false) return;
|
||||
for (const run of nativeRuns) {
|
||||
const mode = Boolean(run.conversationOnly);
|
||||
if (conversationOnlyByRunRef.current.get(run.id) !== mode) cursorByRunRef.current.delete(run.id);
|
||||
if (conversationOnlyByRunRef.current.get(run.id) !== mode) { cursorByRunRef.current.delete(run.id); sequencesRef.current.delete(run.id); }
|
||||
conversationOnlyByRunRef.current.set(run.id, mode);
|
||||
}
|
||||
let cancelled = false;
|
||||
const controller = new AbortController();
|
||||
const timers = new Set<number>();
|
||||
const timers = new Map<string, number>();
|
||||
const pendingRefresh = new Set<string>();
|
||||
const retainedIds = new Set(nativeRuns.map((run) => run.id));
|
||||
const retainMap = <T,>(previous: Map<string, T>) => {
|
||||
const next = new Map([...previous].filter(([id]) => retainedIds.has(id)));
|
||||
@@ -68,13 +112,21 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource
|
||||
return next.size === previous.size ? previous : next;
|
||||
});
|
||||
for (const id of cursorByRunRef.current.keys()) {
|
||||
if (!retainedIds.has(id)) cursorByRunRef.current.delete(id);
|
||||
if (!retainedIds.has(id)) {
|
||||
cursorByRunRef.current.delete(id); sequencesRef.current.delete(id); gapPendingRef.current.delete(id);
|
||||
}
|
||||
}
|
||||
|
||||
const refreshRun = async (run: NativeRunTranscriptSource) => {
|
||||
if (cancelled || historyReadsRef.current.has(run.id)) return;
|
||||
pendingRefresh.delete(run.id);
|
||||
window.clearTimeout(timers.get(run.id));
|
||||
timers.delete(run.id);
|
||||
historyReadsRef.current.set(run.id, controller.signal);
|
||||
let failed = false;
|
||||
try {
|
||||
let cursor = cursorByRunRef.current.get(run.id) ?? 0;
|
||||
const observedBeforeRead = sequence(run.id).observed;
|
||||
const incoming: HeartbeatRunEvent[] = [];
|
||||
for (;;) {
|
||||
const page = await readTranscriptRequest(
|
||||
@@ -93,10 +145,14 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource
|
||||
}
|
||||
// Commit this run's cursor with its rows. A slow sibling must neither
|
||||
// hold its readiness hostage nor stall live polling for this run.
|
||||
cursorByRunRef.current.set(run.id, cursor);
|
||||
const coveredCursor = run.conversationOnly ? Math.max(cursor, observedBeforeRead) : cursor;
|
||||
cursorByRunRef.current.set(run.id, coveredCursor);
|
||||
sequence(run.id).confirm(coveredCursor);
|
||||
failed = sequence(run.id).needsCatchup;
|
||||
if (!failed) gapPendingRef.current.delete(run.id);
|
||||
if (incoming.length > 0) setEventsByRun((previous) => {
|
||||
const next = new Map(previous);
|
||||
next.set(run.id, [...new Map([...(previous.get(run.id) ?? []), ...incoming].map((event) => [event.seq, event])).values()].sort((a, b) => a.seq - b.seq));
|
||||
next.set(run.id, mergeRunEvents(previous.get(run.id) ?? [], incoming));
|
||||
return next;
|
||||
});
|
||||
setErrorsByRun((previous) => {
|
||||
@@ -122,23 +178,34 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource
|
||||
return next;
|
||||
});
|
||||
}
|
||||
if (historyReadsRef.current.get(run.id) === controller.signal) historyReadsRef.current.delete(run.id);
|
||||
if (cancelled) return;
|
||||
setHydratedRunIds((previous) => previous.has(run.id) ? previous : new Set([...previous, run.id]));
|
||||
if (isLive(run.status) || failed) {
|
||||
if (pendingRefresh.has(run.id)) {
|
||||
void refreshRun(run);
|
||||
} else if ((isLive(run.status) && !stream.healthy) || failed) {
|
||||
const timer = window.setTimeout(() => {
|
||||
timers.delete(timer);
|
||||
timers.delete(run.id);
|
||||
void refreshRun(run);
|
||||
}, EVENT_POLL_INTERVAL_MS);
|
||||
timers.add(timer);
|
||||
timers.set(run.id, timer);
|
||||
}
|
||||
};
|
||||
for (const run of nativeRuns) void refreshRun(run);
|
||||
for (const run of nativeRuns) {
|
||||
refreshByRunRef.current.set(run.id, (force = false) => {
|
||||
if (force) pendingRefresh.add(run.id);
|
||||
void refreshRun(run);
|
||||
});
|
||||
void refreshRun(run);
|
||||
}
|
||||
return () => {
|
||||
cancelled = true;
|
||||
refreshByRunRef.current.clear();
|
||||
controller.abort();
|
||||
for (const timer of timers) window.clearTimeout(timer);
|
||||
for (const [id, signal] of historyReadsRef.current) if (signal === controller.signal) historyReadsRef.current.delete(id);
|
||||
for (const timer of timers.values()) window.clearTimeout(timer);
|
||||
};
|
||||
}, [nativeRuns, retryGeneration]);
|
||||
}, [nativeRuns, retryGeneration, visible, options.enabled, options.companyId, stream.healthy, stream.epoch]);
|
||||
|
||||
const transcriptByRun = useMemo(() => {
|
||||
const transcripts = new Map<string, TranscriptEntry[]>();
|
||||
|
||||
Reference in new issue
Block a user