diff --git a/apps/agent-forge/src/observe/useObserve.ts b/apps/agent-forge/src/observe/useObserve.ts index cdd2923cd..001274388 100644 --- a/apps/agent-forge/src/observe/useObserve.ts +++ b/apps/agent-forge/src/observe/useObserve.ts @@ -125,6 +125,24 @@ export function useInsights(filter: InsightsFilter, refreshKey?: unknown): Obser return useObserveQuery(() => observeClient.insights(filter), [JSON.stringify(filter), refreshKey]); } +/** + * {@link useInsights} whose `data` is only ever the answer fetched for the + * current `tag`: after the tag changes (e.g. a new window or granularity) it + * stays undefined (loading) until that tag's answer lands, so nothing draws + * stale rows against the new buckets. Polls under the same tag keep the last answer. + */ +export function useTaggedInsights( + tag: string, + filter: InsightsFilter, + refreshKey?: unknown +): ObserveQuery { + const query = useObserveQuery( + async () => ({ tag, insights: await observeClient.insights(filter) }), + [tag, JSON.stringify(filter), refreshKey] + ); + return { ...query, data: query.data?.tag === tag ? query.data.insights : undefined }; +} + /** * One eval's score rows (`eval=` plus `filter`); no request while * `evalName` is empty. `data` is only ever rows of the current `evalName`: diff --git a/apps/agent-forge/src/screens/monitoring/MetricRow.tsx b/apps/agent-forge/src/screens/monitoring/MetricRow.tsx index 01424bc04..921290ab8 100644 --- a/apps/agent-forge/src/screens/monitoring/MetricRow.tsx +++ b/apps/agent-forge/src/screens/monitoring/MetricRow.tsx @@ -2,28 +2,29 @@ import { fmtCost, fmtDuration, fmtInt, fmtPercent } from '../../observe/format'; import type { ObserveRollupRow } from '../../observe/types'; import { MetricCard } from '../../ui/MetricCard'; import { metricSummary, splitWindow } from './monitoringMetrics'; -import type { MonitoringWindow } from './monitoringSeries'; +import type { ChartWindow, MonitoringWindow } from './monitoringSeries'; const points = (v: number) => `${v.toLocaleString('en-US', { maximumFractionDigits: 1 })} pp`; /** * The four KPI cards over the chart window: runs, error rate, p95 latency and - * cost, each with its per-bucket sparkline and the change against the - * previous window (`rows` reach back one window before `from`). + * cost over the whole window (bucket in progress included), each with its + * per-bucket sparkline and the change over the window's complete buckets + * against as many just before it (`rows` reach back one window before `from`). */ export function MetricRow({ rows, - from, + chart, win, loading, }: { rows: ObserveRollupRow[]; - from: number; + chart: ChartWindow; win: MonitoringWindow; loading: boolean; }) { - const { current, previous } = splitWindow(rows, from); - const m = metricSummary(current, previous); + const split = splitWindow(rows, chart); + const m = metricSummary(split.window, split.compare); const label = `vs previous ${win}`; const delta = (value: number | undefined, goodWhen: 'up' | 'down', format?: (v: number) => string) => value === undefined ? undefined : { value, goodWhen, label, format }; diff --git a/apps/agent-forge/src/screens/monitoring/MonitoringScreen.tsx b/apps/agent-forge/src/screens/monitoring/MonitoringScreen.tsx index c984368f8..51b59a1db 100644 --- a/apps/agent-forge/src/screens/monitoring/MonitoringScreen.tsx +++ b/apps/agent-forge/src/screens/monitoring/MonitoringScreen.tsx @@ -1,7 +1,7 @@ import { useEffect, useMemo, useState } from 'react'; import { AgentSelect, ObserveDaemonDown, ObserveError } from '../../observe/chrome'; -import { useAgents, useDaemonDown, useDaemonHealth, useInsights } from '../../observe/useObserve'; +import { useAgents, useDaemonDown, useDaemonHealth, useInsights, useTaggedInsights } from '../../observe/useObserve'; import { navigate } from '../../router/router'; import { useQueryState } from '../../router/useRoute'; import { ScreenHeader } from '../../shell/ScreenHeader'; @@ -73,7 +73,9 @@ function useMonitoringData(win: MonitoringWindow, agentId: string) { const chart = chartWindow(win, now); const agents = useAgents(undefined, refreshKey); const fleetInsights = useInsights({ days: 1 }, refreshKey); - const insights = useInsights( + // Tagged by window and agent: a switch shows the loading state, not the old rows on the new buckets. + const insights = useTaggedInsights( + `${win}:${agentId}`, { ...insightsSpan(chart), agent_id: agentId || undefined, granularity: chart.granularity }, refreshKey, ); @@ -99,7 +101,7 @@ export function MonitoringScreen() { const labelOf = (id: string) => names.get(id) ?? id.slice(0, 8); const rows = insights.data?.series ?? []; // Charts draw the window only; the earlier rows are the metric cards' comparison. - const windowRows = splitWindow(rows, chart.from).current; + const windowRows = splitWindow(rows, chart).window; const chartsLoading = insights.data === undefined && insights.loading; // An `agent=` id the fleet doesn't list still filters, so the select offers it too. const agentOptions = useMemo(() => { @@ -135,7 +137,7 @@ export function MonitoringScreen() {
{header} - +
{ it('metric cards show the window figures with the change against the previous window', async () => { const w = chartWindow('24h', NOW); - const span = w.to - w.from; daemon([ rollup(w.from, 'a1', 'runs', 10), rollup(w.from, 'a1', 'errors', 1), @@ -190,11 +189,11 @@ describe('MonitoringScreen', () => { rollup(w.from + HOUR, 'a1', 'errors', 2), rollup(w.from + HOUR, 'a1', 'cost_usd', 0.5), rollup(w.from + HOUR, 'a1', 'p95_latency_ms', 1100), - // The previous window: 15 runs, 3 errors, $2, p95 2s. - rollup(w.from - span, 'a1', 'runs', 15), - rollup(w.from - span, 'a1', 'errors', 3), - rollup(w.from - span, 'a1', 'cost_usd', 2), - rollup(w.from - span, 'a1', 'p95_latency_ms', 2000), + // The previous window (the 24 complete buckets before `from`): 15 runs, 3 errors, $2, p95 2s. + rollup(w.from - HOUR, 'a1', 'runs', 15), + rollup(w.from - HOUR, 'a1', 'errors', 3), + rollup(w.from - HOUR, 'a1', 'cost_usd', 2), + rollup(w.from - HOUR, 'a1', 'p95_latency_ms', 2000), ]); const c = render(); await waitFor(() => drawn(c, 'Runs')); @@ -217,6 +216,46 @@ describe('MonitoringScreen', () => { expect(chart(c, 'Runs')!.querySelectorAll('polyline.tsc-line[data-series="ok"]')).toHaveLength(1); }); + it('leaves the bucket in progress out of the deltas, but not out of the values', async () => { + const w = chartWindow('24h', NOW); + const inProgress = w.to - HOUR; + daemon([ + // 10 runs in each side's first complete bucket, plus 5 so far in the bucket holding `now`. + rollup(w.from, 'a1', 'runs', 10), + rollup(inProgress, 'a1', 'runs', 5), + rollup(w.from - HOUR, 'a1', 'runs', 10), + ]); + const c = render(); + await waitFor(() => drawn(c, 'Runs')); + const runs = [...c.querySelectorAll('.metric-card')].find((el) => el.querySelector('.metric-label')?.textContent === 'Runs')!; + expect(runs.querySelector('.metric-value')?.textContent).toBe('15'); + expect(runs.querySelector('.metric-delta')?.textContent).toContain('no change vs previous 24h'); + }); + + it('a window switch shows the cards and charts loading, not the old rows on the new buckets', async () => { + const day = chartWindow('24h', NOW); + const { fetchFn } = daemon([rollup(day.from, 'a1', 'runs', 10), rollup(day.from - HOUR, 'a1', 'runs', 5)]); + const c = render(); + await waitFor(() => drawn(c, 'Runs')); + let release: () => void = () => undefined; + const stalled = new Promise((r) => (release = r)); + const passthrough = fetchFn.getMockImplementation()!; + fetchFn.mockImplementation(async (input) => { + if (String(input).includes('granularity=day')) await stalled; + return passthrough(input); + }); + const button = (label: string) => + [...c.querySelectorAll('.mon-windows button')].find((b) => b.textContent === label) as HTMLButtonElement; + act(() => button('7d').click()); + await waitFor(() => c.querySelector('.mon-windows [aria-pressed="true"]')?.textContent === '7d'); + expect(c.querySelectorAll('.metric-card[aria-busy="true"]').length).toBe(4); + expect(c.querySelector('.metric-delta')).toBeNull(); + expect(drawn(c, 'Runs')).toBe(false); + release(); + await waitFor(() => drawn(c, 'Runs')); + expect(c.querySelectorAll('.metric-card[aria-busy="true"]').length).toBe(0); + }); + it('shows no delta without a previous window', async () => { const w = chartWindow('24h', NOW); daemon([rollup(w.from, 'a1', 'runs', 10)]); diff --git a/apps/agent-forge/src/screens/monitoring/__tests__/monitoringMetrics.test.ts b/apps/agent-forge/src/screens/monitoring/__tests__/monitoringMetrics.test.ts index 2774e835f..a623deafc 100644 --- a/apps/agent-forge/src/screens/monitoring/__tests__/monitoringMetrics.test.ts +++ b/apps/agent-forge/src/screens/monitoring/__tests__/monitoringMetrics.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'; import type { ObserveRollupRow } from '../../../observe/types'; import { metricSummary, splitWindow, windowTotals } from '../monitoringMetrics'; +import { chartWindow, insightsSpan, WINDOWS, type MonitoringWindow } from '../monitoringSeries'; const HOUR = 3_600_000; const T = Date.UTC(2026, 9, 25, 12); @@ -15,11 +16,67 @@ const row = (agent_id: string, metric: string, value: number, bucket_start = T): }); describe('splitWindow', () => { - it('splits rows at the window start: earlier buckets are the previous window', () => { - const rows = [row('a', 'runs', 1, T - HOUR), row('a', 'runs', 2, T), row('a', 'runs', 3, T + HOUR)]; - const { current, previous } = splitWindow(rows, T); - expect(current.map((r) => r.value)).toEqual([2, 3]); - expect(previous.map((r) => r.value)).toEqual([1]); + // A 3-bucket window [T, T + 3h): two complete buckets, then the one still in progress. + const w = { from: T, to: T + 3 * HOUR, bucketMs: HOUR }; + + it('the window is every bucket from `from`; the deltas compare its complete buckets with as many before it', () => { + const rows = [ + row('a', 'runs', 1, T - 3 * HOUR), + row('a', 'runs', 2, T - 2 * HOUR), + row('a', 'runs', 3, T - HOUR), + row('a', 'runs', 4, T), + row('a', 'runs', 5, T + HOUR), + row('a', 'runs', 6, T + 2 * HOUR), + ]; + const split = splitWindow(rows, w); + expect(split.window.map((r) => r.value)).toEqual([4, 5, 6]); + expect(split.compare?.current.map((r) => r.value)).toEqual([4, 5]); + expect(split.compare?.previous.map((r) => r.value)).toEqual([2, 3]); + }); + + it('no comparison while the window has no complete bucket', () => { + const split = splitWindow([row('a', 'runs', 1, T - HOUR), row('a', 'runs', 2, T)], { from: T, to: T + HOUR, bucketMs: HOUR }); + expect(split.window.map((r) => r.value)).toEqual([2]); + expect(split.compare).toBeNull(); + }); +}); + +describe('steady traffic, `now` pinned mid-bucket', () => { + // 120 runs and $0.60 an hour, every hour; the bucket holding `now` has only its elapsed share. + const PER_HOUR = 120; + const steady = (win: MonitoringWindow, now: number) => { + const w = chartWindow(win, now); + const { since, until } = insightsSpan(w); + const rows: ObserveRollupRow[] = []; + for (let b = since; b < until; b += w.bucketMs) { + const hours = (Math.min(b + w.bucketMs, now) - b) / HOUR; + if (hours <= 0) continue; + rows.push(row('a', 'runs', PER_HOUR * hours, b), row('a', 'cost_usd', 0.6 * hours, b)); + } + return { w, rows }; + }; + + for (const win of WINDOWS) { + it(`${win}: runs and cost deltas are 0%`, () => { + const now = Date.UTC(2026, 9, 25, 10, 5); + const { w, rows } = steady(win, now); + const split = splitWindow(rows, w); + const m = metricSummary(split.window, split.compare); + expect(m.runs.delta).toBeCloseTo(0, 9); + expect(m.cost.delta).toBeCloseTo(0, 9); + // The value still counts the bucket in progress. + const visible = rows.filter((r) => r.metric === 'runs' && r.bucket_start >= w.from).reduce((a, r) => a + r.value, 0); + expect(m.runs.value).toBeCloseTo(visible); + }); + } + + it('1h at 10:05: the 5 minutes of the 10:00 bucket count in the value, not in the delta', () => { + const { w, rows } = steady('1h', Date.UTC(2026, 9, 25, 10, 5)); + const split = splitWindow(rows, w); + const m = metricSummary(split.window, split.compare); + expect(m.runs.value).toBeCloseTo(PER_HOUR + PER_HOUR / 12); + expect(m.runs.sparkline).toEqual([PER_HOUR, PER_HOUR / 12]); + expect(m.runs.delta).toBeCloseTo(0, 9); }); }); @@ -66,7 +123,7 @@ describe('metricSummary', () => { ]; it('values come from the current window; sparklines are its per-bucket values', () => { - const m = metricSummary(current, previous); + const m = metricSummary(current, { current, previous }); expect(m.runs.value).toBe(20); expect(m.runs.sparkline).toEqual([10, 10]); expect(m.errorRate.value).toBeCloseTo(0.1); @@ -78,7 +135,7 @@ describe('metricSummary', () => { }); it('deltas: relative % for runs, p95 and cost; percentage points for error rate', () => { - const m = metricSummary(current, previous); + const m = metricSummary(current, { current, previous }); expect(m.runs.delta).toBeCloseTo(100); expect(m.p95.delta).toBeCloseTo(-50); expect(m.cost.delta).toBeCloseTo(-20); @@ -86,19 +143,32 @@ describe('metricSummary', () => { }); it('no delta when the previous window has no rows, or a zero base', () => { - const none = metricSummary(current, []); + const none = metricSummary(current, { current, previous: [] }); expect([none.runs.delta, none.errorRate.delta, none.p95.delta, none.cost.delta]).toEqual([ undefined, undefined, undefined, undefined, ]); - const zero = metricSummary(current, [row('a', 'runs', 0, T - HOUR), row('a', 'cost_usd', 0, T - HOUR)]); + const zero = metricSummary(current, { current, previous: [row('a', 'runs', 0, T - HOUR), row('a', 'cost_usd', 0, T - HOUR)] }); expect([zero.runs.delta, zero.errorRate.delta, zero.cost.delta]).toEqual([undefined, undefined, undefined]); }); + it('no delta at all without a comparison (no complete bucket yet), though the values show', () => { + const m = metricSummary(current, null); + expect(m.runs.value).toBe(20); + expect([m.runs.delta, m.errorRate.delta, m.p95.delta, m.cost.delta]).toEqual([undefined, undefined, undefined, undefined]); + }); + + it('deltas come from the comparison rows, values from the window rows', () => { + const m = metricSummary(current, { current: current.filter((r) => r.bucket_start === T), previous }); + expect(m.runs.value).toBe(20); + expect(m.runs.delta).toBeCloseTo(0); // 10 vs 10 + expect(m.cost.delta).toBeCloseTo(-40); // $0.30 vs $0.50 + }); + it('an empty current window has no error rate or p95 value', () => { - const m = metricSummary([], previous); + const m = metricSummary([], { current: [], previous }); expect(m.runs.value).toBe(0); expect(m.errorRate.value).toBeNull(); expect(m.errorRate.delta).toBeUndefined(); diff --git a/apps/agent-forge/src/screens/monitoring/monitoringMetrics.ts b/apps/agent-forge/src/screens/monitoring/monitoringMetrics.ts index 8b23d107e..36e9b5c9c 100644 --- a/apps/agent-forge/src/screens/monitoring/monitoringMetrics.ts +++ b/apps/agent-forge/src/screens/monitoring/monitoringMetrics.ts @@ -4,8 +4,10 @@ * against the previous window of the same length. * * The screen fetches one rollup query reaching back two window spans and - * splits it at the window start ({@link splitWindow}); the charts and values - * use the current half, the deltas compare against the earlier one. + * splits it ({@link splitWindow}): the charts, values and sparklines use the + * whole window, bucket in progress included; the deltas compare like with + * like - the window's complete buckets against as many complete buckets just + * before it. */ import type { ObserveRollupRow } from '../../observe/types'; @@ -32,9 +34,39 @@ export interface MetricSummary { cost: MetricFigure; } -/** Rows in buckets starting at or after `from` are the current window; earlier ones the previous window. */ -export function splitWindow(rows: ObserveRollupRow[], from: number): { current: ObserveRollupRow[]; previous: ObserveRollupRow[] } { - return { current: rows.filter((r) => r.bucket_start >= from), previous: rows.filter((r) => r.bucket_start < from) }; +/** The bucket grid a window's rows are split on (a `ChartWindow` is one). */ +export interface BucketWindow { + /** Start of the window's first bucket. */ + from: number; + /** End of the bucket holding `now`, the one still in progress. */ + to: number; + bucketMs: number; +} + +/** What the deltas compare: two runs of the same number of complete buckets. */ +export interface WindowComparison { + /** The window's complete buckets: `[from, to - bucketMs)`. */ + current: ObserveRollupRow[]; + /** As many buckets immediately before `from`. */ + previous: ObserveRollupRow[]; +} + +/** + * `window` is every row from `from` on (the charts, values and sparklines); + * `compare` leaves the in-progress bucket out so both sides span the same + * number of complete buckets, and is `null` while the window has none. + */ +export function splitWindow( + rows: ObserveRollupRow[], + w: BucketWindow, +): { window: ObserveRollupRow[]; compare: WindowComparison | null } { + const completeEnd = w.to - w.bucketMs; + const prevStart = w.from - (completeEnd - w.from); + const within = (start: number, end: number) => rows.filter((r) => r.bucket_start >= start && r.bucket_start < end); + return { + window: rows.filter((r) => r.bucket_start >= w.from), + compare: completeEnd > w.from ? { current: within(w.from, completeEnd), previous: within(prevStart, w.from) } : null, + }; } const sum = (rows: ObserveRollupRow[], metric: string) => @@ -80,25 +112,39 @@ function perBucket(rows: ObserveRollupRow[]): WindowTotals[] { const present = (values: (number | null)[]): number[] => values.filter((v): v is number => v !== null); -/** - * The four Monitoring metric cards. Runs, p95 and cost deltas are relative % - * changes; the error-rate delta is in percentage points. A previous window - * with no runs (or no p95 / cost) gives no delta. - */ -export function metricSummary(current: ObserveRollupRow[], previous: ObserveRollupRow[]): MetricSummary { - const now = windowTotals(current); - const before = windowTotals(previous); - const buckets = perBucket(current); +/** The four deltas of `compare` (all undefined without one). */ +function deltas(compare: WindowComparison | null) { + if (compare === null) return { runs: undefined, errorRate: undefined, p95: undefined, cost: undefined }; + const now = windowTotals(compare.current); + const before = windowTotals(compare.previous); const rate = errorRate(now); const prevRate = errorRate(before); return { - runs: { value: now.runs, delta: relative(now.runs, before.runs), sparkline: buckets.map((b) => b.runs) }, + runs: relative(now.runs, before.runs), + errorRate: rate === null || prevRate === null ? undefined : (rate - prevRate) * 100, + p95: relative(now.p95, before.p95), + cost: relative(now.cost, before.cost), + }; +} + +/** + * The four Monitoring metric cards: values and sparklines over `window`, + * deltas over `compare` (see {@link splitWindow}). Runs, p95 and cost deltas + * are relative % changes; the error-rate delta is in percentage points. No + * comparison, or a previous side with no runs (or no p95 / cost), gives no delta. + */ +export function metricSummary(window: ObserveRollupRow[], compare: WindowComparison | null): MetricSummary { + const now = windowTotals(window); + const buckets = perBucket(window); + const d = deltas(compare); + return { + runs: { value: now.runs, delta: d.runs, sparkline: buckets.map((b) => b.runs) }, errorRate: { - value: rate, - delta: rate === null || prevRate === null ? undefined : (rate - prevRate) * 100, + value: errorRate(now), + delta: d.errorRate, sparkline: present(buckets.map((b) => (b.runs > 0 ? (b.errors * 100) / b.runs : null))), }, - p95: { value: now.p95, delta: relative(now.p95, before.p95), sparkline: present(buckets.map((b) => b.p95)) }, - cost: { value: now.cost, delta: relative(now.cost, before.cost), sparkline: buckets.map((b) => b.cost) }, + p95: { value: now.p95, delta: d.p95, sparkline: present(buckets.map((b) => b.p95)) }, + cost: { value: now.cost, delta: d.cost, sparkline: buckets.map((b) => b.cost) }, }; } diff --git a/apps/agent-forge/src/screens/monitoring/monitoringSeries.ts b/apps/agent-forge/src/screens/monitoring/monitoringSeries.ts index e7549c7d4..b9171807f 100644 --- a/apps/agent-forge/src/screens/monitoring/monitoringSeries.ts +++ b/apps/agent-forge/src/screens/monitoring/monitoringSeries.ts @@ -54,7 +54,9 @@ export function chartWindow(window: MonitoringWindow, now: number): ChartWindow /** * The rollup query behind the charts and metric cards: the window plus the one - * before it (same length), so the cards can show a change without a second request. + * before it (same length), so the cards can show a change without a second + * request. The deltas need one bucket less than that (they leave the bucket in + * progress out of both sides; see `splitWindow`). */ export function insightsSpan(w: ChartWindow): { since: number; until: number } { return { since: w.from - (w.to - w.from), until: w.to };