feat(FN-5416): instrument SSE stream resume events across PR checks, dev-se

Adds structured stream-resume instrumentation to the dashboard across four SSE hooks (PR checks, dev-server logs, research, background sessions), wiring `[wake-trigger-diagnostics]` log markers and view-resume route guards so the engine can distinguish cold starts from resume paths; includes tests f

Fusion-Task-Id: FN-5416
This commit is contained in:
Fusion (runfusion.ai)
2026-05-21 18:14:47 -07:00
committed by gsxdsm
parent 921e0e616f
commit 854045f430
12 changed files with 466 additions and 5 deletions

View File

@@ -8,6 +8,7 @@ import { usePreviewEmbed } from "../hooks/usePreviewEmbed";
import type { ToastType } from "../hooks/useToast";
import { DevServerLogViewer } from "./DevServerLogViewer";
import { PreviewIframe } from "./PreviewIframe";
import { recordResumeEvent } from "../utils/resumeInstrumentation";
interface DevServerViewProps {
addToast: (msg: string, type?: ToastType) => void;
@@ -29,6 +30,8 @@ const STATUS_BADGE_CONFIG: Record<"stopped" | "starting" | "running" | "failed"
failed: { className: "dev-server-status-badge--failed", label: "Failed" },
};
let devServerViewWasPreviouslyInactive = false;
function normalizeError(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
@@ -79,6 +82,26 @@ function truncateCommand(command: string): string {
}
export function DevServerView({ addToast, projectId }: DevServerViewProps) {
useEffect(() => {
recordResumeEvent({
view: "DevServerView",
trigger: devServerViewWasPreviouslyInactive ? "route-active" : "remount",
projectId,
replayAttempted: false,
});
devServerViewWasPreviouslyInactive = false;
return () => {
devServerViewWasPreviouslyInactive = true;
recordResumeEvent({
view: "DevServerView",
trigger: "route-inactive",
projectId,
replayAttempted: false,
});
};
}, [projectId]);
const {
session,
detectedCommands,

View File

@@ -7,6 +7,7 @@ import type { ResearchProviderOption } from "../research-types";
import { ResearchTaskActionModal } from "./ResearchTaskActionModal";
import type { SectionId } from "./SettingsModal";
import "./ResearchView.css";
import { recordResumeEvent } from "../utils/resumeInstrumentation";
interface ResearchViewProps {
projectId?: string;
@@ -33,7 +34,28 @@ const PROVIDER_LABELS: Record<ResearchProviderOption, string> = {
"llm-synthesis": "LLM Synthesis",
};
let researchViewWasPreviouslyInactive = false;
export function ResearchView({ projectId, addToast, onOpenSettings, readinessVersion = 0 }: ResearchViewProps) {
useEffect(() => {
recordResumeEvent({
view: "ResearchView",
trigger: researchViewWasPreviouslyInactive ? "route-active" : "remount",
projectId,
replayAttempted: false,
});
researchViewWasPreviouslyInactive = false;
return () => {
researchViewWasPreviouslyInactive = true;
recordResumeEvent({
view: "ResearchView",
trigger: "route-inactive",
projectId,
replayAttempted: false,
});
};
}, [projectId]);
const {
runs,
selectedRun,

View File

@@ -0,0 +1,73 @@
import { act, renderHook, waitFor } from "@testing-library/react";
import { beforeEach, describe, expect, it, vi } from "vitest";
const recordResumeEvent = vi.fn();
const subscribeSse = vi.fn();
const requestSync = vi.fn();
const broadcastUpdate = vi.fn();
const broadcastCompleted = vi.fn();
vi.mock("../../utils/resumeInstrumentation", () => ({
recordResumeEvent,
}));
vi.mock("../../sse-bus", () => ({
subscribeSse,
}));
vi.mock("../useAiSessionSync", () => ({
useAiSessionSync: () => ({
sessions: new Map(),
requestSync,
broadcastUpdate,
broadcastCompleted,
}),
}));
vi.mock("../../api", () => ({
fetchAiSessions: vi.fn().mockResolvedValue([]),
deleteAiSession: vi.fn(),
cancelPlanning: vi.fn(),
cancelSubtaskBreakdown: vi.fn(),
cancelMissionInterview: vi.fn(),
}));
describe("useBackgroundSessions resume instrumentation", () => {
beforeEach(() => {
recordResumeEvent.mockReset();
subscribeSse.mockReset();
requestSync.mockReset();
});
it("emits sse-open and reconnect instrumentation, then refreshes", async () => {
const { fetchAiSessions } = await import("../../api");
const mockFetch = vi.mocked(fetchAiSessions);
let reconnect: (() => void) | undefined;
subscribeSse.mockImplementation((_url, handlers) => {
reconnect = handlers.onReconnect;
return () => {};
});
const { useBackgroundSessions } = await import("../useBackgroundSessions");
renderHook(() => useBackgroundSessions("proj-1"));
await waitFor(() => {
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useBackgroundSessions",
trigger: "sse-open",
sseChannel: "/api/events?projectId=proj-1",
}));
});
mockFetch.mockClear();
act(() => reconnect?.());
await waitFor(() => expect(mockFetch).toHaveBeenCalled());
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useBackgroundSessions",
trigger: "sse-reconnect",
sseChannel: "/api/events?projectId=proj-1",
}));
});
});

View File

@@ -0,0 +1,78 @@
import { act, renderHook, waitFor } from "@testing-library/react";
import { beforeEach, describe, expect, it, vi } from "vitest";
const recordResumeEvent = vi.fn();
const subscribeSse = vi.fn();
vi.mock("../../utils/resumeInstrumentation", () => ({
recordResumeEvent,
}));
vi.mock("../../sse-bus", () => ({
subscribeSse,
}));
vi.mock("../../api", () => ({
fetchDevServerLogHistory: vi.fn(),
getDevServerLogsStreamUrl: vi.fn((projectId?: string) => `/api/dev-server/logs/stream${projectId ? `?projectId=${projectId}` : ""}`),
}));
describe("useDevServerLogs resume instrumentation", () => {
beforeEach(() => {
recordResumeEvent.mockReset();
subscribeSse.mockReset();
});
it("emits sse-open and sse-reconnect with replay event id", async () => {
const { fetchDevServerLogHistory } = await import("../../api");
vi.mocked(fetchDevServerLogHistory)
.mockResolvedValueOnce({ lines: [{ id: 5, text: "a", stream: "stdout", timestamp: "" }], totalLines: 1 })
.mockResolvedValueOnce({ lines: [], totalLines: 1 });
let reconnect: (() => void) | undefined;
subscribeSse.mockImplementation((_url, handlers) => {
reconnect = handlers.onReconnect;
return () => {};
});
const { useDevServerLogs } = await import("../useDevServerLogs");
renderHook(({ projectId }) => useDevServerLogs(projectId, true), { initialProps: { projectId: "proj-1" } });
await waitFor(() => {
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useDevServerLogs",
trigger: "sse-open",
sseChannel: "/api/dev-server/logs/stream?projectId=proj-1",
}));
});
act(() => reconnect?.());
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useDevServerLogs",
trigger: "sse-reconnect",
replayFromEventId: 5,
replayAttempted: true,
}));
});
it("emits project-context-change when project id changes", async () => {
const { fetchDevServerLogHistory } = await import("../../api");
vi.mocked(fetchDevServerLogHistory).mockResolvedValue({ lines: [], totalLines: 0 });
subscribeSse.mockImplementation(() => () => {});
const { useDevServerLogs } = await import("../useDevServerLogs");
const { rerender } = renderHook(({ projectId }) => useDevServerLogs(projectId, true), {
initialProps: { projectId: "proj-1" },
});
rerender({ projectId: "proj-2" });
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useDevServerLogs",
trigger: "project-context-change",
projectId: "proj-2",
reason: "context-version-bumped",
}));
});
});

View File

@@ -0,0 +1,112 @@
import { act, renderHook, waitFor } from "@testing-library/react";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
const recordResumeEvent = vi.fn();
vi.mock("../../utils/resumeInstrumentation", () => ({
recordResumeEvent,
}));
vi.mock("../../api", () => ({
fetchPrChecks: vi.fn(),
}));
async function flushPromises(): Promise<void> {
await Promise.resolve();
await Promise.resolve();
}
describe("usePrChecksStream resume instrumentation", () => {
beforeEach(() => {
recordResumeEvent.mockReset();
});
afterEach(() => {
vi.useRealTimers();
});
it("emits remount and visibility resume events with task detail", async () => {
const { fetchPrChecks } = await import("../../api");
const mockFetchPrChecks = vi.mocked(fetchPrChecks);
mockFetchPrChecks.mockResolvedValue({ checks: [], rollup: "pending", lastCheckedAt: "" });
const { usePrChecksStream } = await import("../usePrChecksStream");
renderHook(() => usePrChecksStream({ taskId: "FN-1", projectId: "proj-1", prNumber: 42, enabled: true }));
await waitFor(() => {
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "usePrChecksStream",
trigger: "remount",
detail: { taskId: "FN-1", prNumber: 42 },
}));
});
Object.defineProperty(document, "hidden", { configurable: true, value: true });
act(() => {
document.dispatchEvent(new Event("visibilitychange"));
});
Object.defineProperty(document, "hidden", { configurable: true, value: false });
act(() => {
document.dispatchEvent(new Event("visibilitychange"));
});
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "usePrChecksStream",
trigger: "visibility",
reason: "hidden",
detail: { taskId: "FN-1", prNumber: 42 },
}));
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "usePrChecksStream",
trigger: "visibility",
reason: "visible-resume",
detail: { taskId: "FN-1", prNumber: 42 },
}));
});
it("keeps stable-check polling backoff behavior", async () => {
vi.useFakeTimers();
const { fetchPrChecks } = await import("../../api");
const mockFetchPrChecks = vi.mocked(fetchPrChecks);
mockFetchPrChecks.mockResolvedValue({
checks: [{ name: "ci", state: "success" }],
rollup: "pending",
lastCheckedAt: "",
});
const { usePrChecksStream } = await import("../usePrChecksStream");
renderHook(() => usePrChecksStream({ taskId: "FN-1", projectId: "proj-1", prNumber: 42, enabled: true }));
await act(async () => {
await flushPromises();
});
mockFetchPrChecks.mockClear();
await act(async () => {
vi.advanceTimersByTime(15_000);
await flushPromises();
});
await act(async () => {
vi.advanceTimersByTime(15_000);
await flushPromises();
});
await act(async () => {
vi.advanceTimersByTime(15_000);
await flushPromises();
});
expect(mockFetchPrChecks).toHaveBeenCalledTimes(3);
await act(async () => {
vi.advanceTimersByTime(59_000);
await flushPromises();
});
expect(mockFetchPrChecks).toHaveBeenCalledTimes(3);
await act(async () => {
vi.advanceTimersByTime(1_000);
await flushPromises();
});
expect(mockFetchPrChecks).toHaveBeenCalledTimes(4);
});
});

View File

@@ -0,0 +1,63 @@
import { act, renderHook, waitFor } from "@testing-library/react";
import { beforeEach, describe, expect, it, vi } from "vitest";
const recordResumeEvent = vi.fn();
const subscribeSse = vi.fn();
vi.mock("../../utils/resumeInstrumentation", () => ({
recordResumeEvent,
}));
vi.mock("../../sse-bus", () => ({
subscribeSse,
}));
vi.mock("../../api", () => ({
listResearchRuns: vi.fn().mockResolvedValue({ runs: [], availability: { available: true } }),
getResearchRun: vi.fn(),
createResearchRun: vi.fn(),
cancelResearchRun: vi.fn(),
retryResearchRun: vi.fn(),
exportResearchRun: vi.fn(),
createTaskFromResearchRun: vi.fn(),
attachResearchRunToTask: vi.fn(),
}));
describe("useResearch resume instrumentation", () => {
beforeEach(() => {
recordResumeEvent.mockReset();
subscribeSse.mockReset();
});
it("emits sse-open and reconnect instrumentation, then refreshes", async () => {
const { listResearchRuns } = await import("../../api");
const mockList = vi.mocked(listResearchRuns);
let reconnect: (() => void) | undefined;
subscribeSse.mockImplementation((_url, handlers) => {
reconnect = handlers.onReconnect;
return () => {};
});
const { useResearch } = await import("../useResearch");
renderHook(() => useResearch({ projectId: "proj-1" }));
await waitFor(() => {
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useResearch",
trigger: "sse-open",
sseChannel: "/api/events?projectId=proj-1",
}));
});
mockList.mockClear();
act(() => reconnect?.());
await waitFor(() => expect(mockList).toHaveBeenCalled());
expect(recordResumeEvent).toHaveBeenCalledWith(expect.objectContaining({
view: "useResearch",
trigger: "sse-reconnect",
sseChannel: "/api/events?projectId=proj-1",
}));
});
});

View File

@@ -10,6 +10,7 @@ import {
import { useAiSessionSync } from "./useAiSessionSync";
import { getSessionTabId } from "../utils/getSessionTabId";
import { subscribeSse } from "../sse-bus";
import { recordResumeEvent } from "../utils/resumeInstrumentation";
interface UseBackgroundSessionsResult {
sessions: AiSessionSummary[];
@@ -273,7 +274,9 @@ export function useBackgroundSessions(projectId?: string): UseBackgroundSessions
}
};
return subscribeSse(`/api/events${params}`, {
const sseChannel = `/api/events${params}`;
const unsubscribe = subscribeSse(sseChannel, {
events: {
"ai_session:updated": handleUpdated,
"ai_session:deleted": handleDeleted,
@@ -283,9 +286,25 @@ export function useBackgroundSessions(projectId?: string): UseBackgroundSessions
// the channel was down stay invisible to this hook and the AI pill
// count gets stuck on stale sessions.
onReconnect: () => {
recordResumeEvent({
view: "useBackgroundSessions",
trigger: "sse-reconnect",
projectId,
replayAttempted: false,
sseChannel,
});
refresh();
},
});
recordResumeEvent({
view: "useBackgroundSessions",
trigger: "sse-open",
projectId,
replayAttempted: false,
sseChannel,
});
return unsubscribe;
}, [broadcastCompleted, broadcastUpdate, projectId, refresh]);
const dismissSession = useCallback(async (id: string) => {

View File

@@ -5,6 +5,7 @@ import {
type DevServerLogHistoryEntry,
} from "../api";
import { subscribeSse } from "../sse-bus";
import { recordResumeEvent } from "../utils/resumeInstrumentation";
export interface DevServerLogEntry {
id: number;
@@ -106,6 +107,13 @@ export function useDevServerLogs(projectId: string | undefined, enabled: boolean
previousProjectIdRef.current = projectId;
previousEnabledRef.current = enabled;
projectContextVersionRef.current++;
recordResumeEvent({
view: "useDevServerLogs",
trigger: "project-context-change",
projectId,
replayAttempted: false,
reason: "context-version-bumped",
});
cancelledRef.current = true;
lastSeenEventIdRef.current = 0;
nextSyntheticIdRef.current = 1;
@@ -268,6 +276,15 @@ export function useDevServerLogs(projectId: string | undefined, enabled: boolean
},
},
onReconnect: () => {
recordResumeEvent({
view: "useDevServerLogs",
trigger: "sse-reconnect",
projectId,
replayAttempted: true,
replayFromEventId: lastSeenEventIdRef.current ?? null,
sseChannel: streamUrl,
reason: "history-replay",
});
if (cancelledRef.current || projectContextVersionRef.current !== contextVersionAtStart) {
return;
}
@@ -299,6 +316,13 @@ export function useDevServerLogs(projectId: string | undefined, enabled: boolean
});
},
});
recordResumeEvent({
view: "useDevServerLogs",
trigger: "sse-open",
projectId,
replayAttempted: false,
sseChannel: streamUrl,
});
}
void init();

View File

@@ -1,5 +1,6 @@
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { fetchPrChecks, type PrCheckStatus, type PrChecksResponse } from "../api";
import { recordResumeEvent } from "../utils/resumeInstrumentation";
type RollupState = PrChecksResponse["rollup"];
@@ -152,9 +153,24 @@ export function usePrChecksStream({
return;
}
recordResumeEvent({
view: "usePrChecksStream",
trigger: "remount",
projectId,
replayAttempted: false,
detail: { taskId, prNumber },
});
void poll();
const onVisibilityChange = () => {
recordResumeEvent({
view: "usePrChecksStream",
trigger: "visibility",
projectId,
replayAttempted: false,
reason: document.hidden ? "hidden" : "visible-resume",
detail: { taskId, prNumber },
});
if (document.hidden) {
clearTimer();
abortRef.current?.abort();
@@ -169,7 +185,7 @@ export function usePrChecksStream({
clearTimer();
abortRef.current?.abort();
};
}, [clearTimer, poll, shouldPoll]);
}, [clearTimer, poll, prNumber, projectId, shouldPoll, taskId]);
return useMemo(() => ({
checks,

View File

@@ -15,6 +15,7 @@ import {
type ResearchActionErrorCode,
} from "../api";
import { subscribeSse } from "../sse-bus";
import { recordResumeEvent } from "../utils/resumeInstrumentation";
import type { ResearchAvailability, ResearchRunDetail, ResearchRunListItem } from "../research-types";
import { readCache, SWR_CACHE_KEYS, SWR_DEFAULT_MAX_AGE_MS, SWR_LONG_MAX_AGE_MS, writeCache } from "../utils/swrCache";
@@ -211,7 +212,8 @@ export function useResearch(options?: { projectId?: string }) {
}
};
const unsubscribe = subscribeSse(`/api/events${query}`, {
const sseChannel = `/api/events${query}`;
const unsubscribe = subscribeSse(sseChannel, {
events: {
"research:run:created": refreshIfActive,
"research:run:updated": refreshIfActive,
@@ -219,7 +221,23 @@ export function useResearch(options?: { projectId?: string }) {
"research:run:failed": refreshIfActive,
"research:run:cancelled": refreshIfActive,
},
onReconnect: refreshIfActive,
onReconnect: () => {
recordResumeEvent({
view: "useResearch",
trigger: "sse-reconnect",
projectId,
replayAttempted: false,
sseChannel,
});
refreshIfActive();
},
});
recordResumeEvent({
view: "useResearch",
trigger: "sse-open",
projectId,
replayAttempted: false,
sseChannel,
});
const pollTimer = window.setInterval(refreshIfActive, POLL_INTERVAL_MS);

View File

@@ -15,7 +15,7 @@ const qualityAppTests = [
"app/components/__tests__/{ActiveAgentsPanel,ActivityLogModal,AgentMentionPopup,AgentMetricsBar,AgentOnboardingModal,AgentReflectionsTab,AgentTokenStatsPanel,App,AuthTokenRecoveryDialog,Board,board-mobile,board-mobile-view-switch,ChatView,ChatView.autosize,ChatView.chat-input-autosize,ChatView.default-model-icon,ChatView.draft,ChatView.hash-mention,ChatView.rooms,ChatView.scroll-to-top,ChatView.swipe-back,Column,ConfirmDialog,ConversationHistory,DashboardLoader,DevServerView.mobile,DirectoryPicker,DuplicateWarningModal,ErrorBoundary,ExecutorStatusBar,FileBrowser,FileEditor,GitHubBadge,InlineCreateCard,LoginInstructions,MemoryView,MergeAdvanceNotice,MessageComposer,MessageComposer.autosize,MobileNavBar,NewTaskModal,NewTaskModal.shared-cache,NodeCard,NodeHealthDot,NodeStatusIndicator,PlanningModeModal.autosize,PrChecksList,PrCreateModal,PrCreateModal.layout,ProjectCard,ProjectSelector,ProviderIcon,PrPanel,PrPanel.merge,PrPanel.reviews,QuickChatFAB,QuickChatFAB.shared-cache,ReliabilityView,ResearchView,SecretsView,SecretsView.mobile,SettingsModal,SettingsModal.worktrunk,StashConflictModal,StashRecoveryView,TaskCard,TaskCard.badge-height,TaskCard.badge-wrap,TaskCard.footer-wrap,TaskChangesTab,TaskComments,TaskDetailModal,TaskDetailModal.create-pr-e2e,TaskDetailModal.create-pr-integration,TaskDetailModal.github-tracking-header,TaskDetailModal.github-tracking-stale,TaskDetailModal.rebind-banner,TaskDocumentsTab,TaskForm,TaskIdIntegrityBanner,TrackingRepoSelect,WorkflowResultsTab,WorktrunkInstallApprovalDetails}.test.tsx",
// Hooks and utilities are fast, user-visible state/formatting behavior.
"app/context/**/*.test.tsx",
"app/hooks/__tests__/{useAgents,useAgentLogs,useAppSettings,useAuthOnboarding,useConfirm,useCurrentProject,useNodes,useNodes.resume-instrumentation,useNodeSettingsSync,useProjects,useProjects.resume-instrumentation,useMeshState.resume-instrumentation,useManagedDockerNodes.resume-instrumentation,useQuickChat,useTasks,useTasks.resume-instrumentation,useChatRooms.resume-instrumentation,useTerminalSessions,useTheme,useToast,useUsageData,useViewState}.test.{ts,tsx}",
"app/hooks/__tests__/{useAgents,useAgentLogs,useAppSettings,useAuthOnboarding,useConfirm,useCurrentProject,useNodes,useNodes.resume-instrumentation,useNodeSettingsSync,useProjects,useProjects.resume-instrumentation,useMeshState.resume-instrumentation,useManagedDockerNodes.resume-instrumentation,usePrChecksStream.resume-instrumentation,useDevServerLogs.resume-instrumentation,useResearch.resume-instrumentation,useBackgroundSessions.resume-instrumentation,useQuickChat,useTasks,useTasks.resume-instrumentation,useChatRooms.resume-instrumentation,useTerminalSessions,useTheme,useToast,useUsageData,useViewState}.test.{ts,tsx}",
"app/utils/**/*.test.{ts,tsx}",
];