From 88632788ece77852d26ab39af4aefe5aa20c3bec Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Sun, 30 Aug 2026 18:59:32 -0300 Subject: [PATCH] =?UTF-8?q?feat(dashboard):=20useOrchestrationSnapshot=20?= =?UTF-8?q?=E2=80=94=20allSettled=20polling=20+=20WS=20requests=20trigger?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../hooks/useOrchestrationSnapshot.ts | 123 ++++++++++++++++++ .../unit/ui/useOrchestrationSnapshot.test.tsx | 82 ++++++++++++ 2 files changed, 205 insertions(+) create mode 100644 src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts create mode 100644 tests/unit/ui/useOrchestrationSnapshot.test.tsx diff --git a/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts b/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts new file mode 100644 index 0000000000..759d6ba831 --- /dev/null +++ b/src/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot.ts @@ -0,0 +1,123 @@ +"use client"; +/** Polls the 3 agent sources (allSettled), listens to the `requests` WS channel as a refetch trigger. */ +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { useLiveDashboard } from "@/hooks/useLiveDashboard"; +import type { CloudAgentTask } from "@/lib/cloudAgent/types"; +import type { A2ATask } from "@/lib/a2a/taskManager"; +import type { FleetSnapshot } from "@/lib/conductor/hubProxy"; +import { fromCloudAgent } from "../model/fromCloudAgent"; +import { fromA2A } from "../model/fromA2A"; +import { fromConductor } from "../model/fromConductor"; +import { mergeSnapshot } from "../model/mergeSnapshot"; +import type { OrchSnapshot, SourceStatus } from "../model/orchestrationTypes"; + +export const POLL_MS = 5_000; +export const WS_REFETCH_DEBOUNCE_MS = 1_000; + +interface Raw { + cloudAgent: CloudAgentTask[]; + a2a: A2ATask[]; + conductor: FleetSnapshot; +} +const EMPTY_RAW: Raw = { cloudAgent: [], a2a: [], conductor: { offline: true, runners: [], tasks: [] } }; + +async function fetchJson(url: string, signal: AbortSignal): Promise { + const res = await fetch(url, { signal, cache: "no-store" }); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + return res.json() as Promise; +} + +export function useOrchestrationSnapshot() { + // `raw` and `polledAt` are React state (not refs) so the merge below reads them + // during render like any other state — a ref read during render trips the + // `react-hooks/refs` lint rule, and computing `Date.now()` inline in the memo + // factory trips `react-hooks/purity`. Sampling `Date.now()` once per poll (inside + // the effect, not during render) keeps `mergeSnapshot`'s staleness math correct + // without either violation. + const [raw, setRaw] = useState(EMPTY_RAW); + const [statuses, setStatuses] = useState([]); + const [polledAt, setPolledAt] = useState(0); + const [isLoading, setIsLoading] = useState(true); + const [showCompleted, setShowCompleted] = useState(false); + const debounceRef = useRef | null>(null); + // Populated by the mount effect below; lets `refetch()` (and the WS debounce + // trigger) reach the same poll loop without hoisting it out of the effect — + // hoisting to a top-level `useCallback` invoked from the effect body trips + // `react-hooks/set-state-in-effect`. + const pollRef = useRef<() => void>(() => {}); + + useEffect(() => { + const controller = new AbortController(); + + const poll = async () => { + const [ca, a2a, cond] = await Promise.allSettled([ + fetchJson<{ data: CloudAgentTask[] }>("/api/v1/agents/tasks?limit=100", controller.signal), + fetchJson<{ tasks: A2ATask[] }>("/api/a2a/tasks?limit=200", controller.signal), + fetchJson("/api/conductor/fleet", controller.signal), + ]); + if (controller.signal.aborted) return; + const nowMs = Date.now(); + const nowIso = new Date(nowMs).toISOString(); + const next: SourceStatus[] = []; + if (ca.status === "fulfilled") next.push({ source: "cloud-agent", ok: true }); + else next.push({ source: "cloud-agent", ok: false, error: String(ca.reason), staleSince: nowIso }); + if (a2a.status === "fulfilled") next.push({ source: "a2a", ok: true }); + else next.push({ source: "a2a", ok: false, error: String(a2a.reason), staleSince: nowIso }); + if (cond.status === "fulfilled") { + next.push({ source: "conductor", ok: true, offline: cond.value.offline }); + } else next.push({ source: "conductor", ok: false, error: String(cond.reason), staleSince: nowIso }); + + // Failed sources keep the previously stored slice — only overwrite what + // actually resolved this round ("last good data" contract from the brief). + setRaw((prev) => ({ + cloudAgent: ca.status === "fulfilled" ? ca.value.data : prev.cloudAgent, + a2a: a2a.status === "fulfilled" ? a2a.value.tasks : prev.a2a, + conductor: cond.status === "fulfilled" ? cond.value : prev.conductor, + })); + setStatuses(next); + setPolledAt(nowMs); + setIsLoading(false); + }; + + pollRef.current = () => void poll(); + void poll(); + const id = setInterval(() => void poll(), POLL_MS); + return () => { + clearInterval(id); + controller.abort(); + if (debounceRef.current) clearTimeout(debounceRef.current); + }; + }, []); + + const refetch = useCallback(() => { + pollRef.current(); + }, []); + + useLiveDashboard({ + channels: ["requests"], + onEvent: (payload) => { + if (payload.channel !== "requests") return; + if (debounceRef.current) return; // debounce burst → one refetch + debounceRef.current = setTimeout(() => { + debounceRef.current = null; + refetch(); + }, WS_REFETCH_DEBOUNCE_MS); + }, + }); + + const snapshot: OrchSnapshot = useMemo( + () => + mergeSnapshot( + { + cloudAgent: fromCloudAgent(raw.cloudAgent), + a2a: fromA2A(raw.a2a), + conductor: fromConductor(raw.conductor), + }, + statuses, + { now: polledAt, showCompleted } + ), + [raw, statuses, showCompleted, polledAt] + ); + + return { snapshot, isLoading, showCompleted, setShowCompleted, refetch }; +} diff --git a/tests/unit/ui/useOrchestrationSnapshot.test.tsx b/tests/unit/ui/useOrchestrationSnapshot.test.tsx new file mode 100644 index 0000000000..f79f95f624 --- /dev/null +++ b/tests/unit/ui/useOrchestrationSnapshot.test.tsx @@ -0,0 +1,82 @@ +// @vitest-environment jsdom +import React, { act } from "react"; +import { createRoot } from "react-dom/client"; +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; + +// Capture the onEvent handler the hook registers on the requests channel. +let capturedOnEvent: ((p: { channel: string }) => void) | null = null; +vi.mock("@/hooks/useLiveDashboard", () => ({ + useLiveDashboard: (opts: { onEvent?: (p: { channel: string }) => void }) => { + capturedOnEvent = opts.onEvent ?? null; + return { connection: { isConnected: true }, events: [] }; + }, +})); + +import { useOrchestrationSnapshot } from "@/app/(dashboard)/dashboard/orchestration/hooks/useOrchestrationSnapshot"; + +function HookProbe({ onRender }: { onRender: (v: ReturnType) => void }) { + onRender(useOrchestrationSnapshot()); + return null; +} + +const okJson = (body: unknown) => Promise.resolve({ ok: true, json: () => Promise.resolve(body) } as Response); + +describe("useOrchestrationSnapshot", () => { + let container: HTMLDivElement; + let root: ReturnType; + beforeEach(() => { + vi.useFakeTimers(); + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + }); + afterEach(() => { + act(() => root.unmount()); + container.remove(); + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("polls the three endpoints and builds a snapshot; a failed source keeps the last good data", async () => { + let latest: ReturnType | null = null; + const task = { id: "t1", providerId: "devin", status: "running", prompt: "p", source: { repoName: "r", repoUrl: "https://x" }, options: {}, activities: [], createdAt: "2026-08-30T10:00:00Z", updatedAt: "2026-08-30T10:00:00Z" }; + const fetchMock = vi.fn((url: string) => { + if (url.startsWith("/api/v1/agents/tasks")) return okJson({ data: [task] }); + if (url.startsWith("/api/a2a/tasks")) return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + return okJson({ offline: false, runners: [], tasks: [] }); + }); + vi.stubGlobal("fetch", fetchMock); + + await act(async () => { root.render( { latest = v; }} />); }); + await act(async () => { await vi.advanceTimersByTimeAsync(10); }); + expect(latest!.snapshot.nodes.some((n) => n.id === "cloud-agent:t1")).toBe(true); + + // Second tick: cloud agent fails — node must survive from the last good photo, source marked stale. + fetchMock.mockImplementation((url: string) => { + if (url.startsWith("/api/v1/agents/tasks")) return Promise.reject(new Error("boom")); + if (url.startsWith("/api/a2a/tasks")) return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + return okJson({ offline: false, runners: [], tasks: [] }); + }); + await act(async () => { await vi.advanceTimersByTimeAsync(5_100); }); + expect(latest!.snapshot.nodes.some((n) => n.id === "cloud-agent:t1")).toBe(true); + const st = latest!.snapshot.sources.find((s) => s.source === "cloud-agent"); + expect(st?.ok).toBe(false); + }); + + it("a requests-channel WS event triggers a debounced immediate refetch", async () => { + const fetchMock = vi.fn((url: string) => { + if (url.startsWith("/api/v1/agents/tasks")) return okJson({ data: [] }); + if (url.startsWith("/api/a2a/tasks")) return okJson({ tasks: [], total: 0, limit: 200, offset: 0 }); + return okJson({ offline: false, runners: [], tasks: [] }); + }); + vi.stubGlobal("fetch", fetchMock); + await act(async () => { root.render( {}} />); }); + await act(async () => { await vi.advanceTimersByTimeAsync(10); }); + const callsAfterMount = fetchMock.mock.calls.length; + + act(() => { capturedOnEvent?.({ channel: "requests" }); capturedOnEvent?.({ channel: "requests" }); }); + await act(async () => { await vi.advanceTimersByTimeAsync(1_100); }); + // Two burst events → exactly ONE extra round of 3 fetches (debounce), not two. + expect(fetchMock.mock.calls.length).toBe(callsAfterMount + 3); + }); +});