Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions apps/agent-forge/src/observe/useObserve.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<ObserveInsights> {
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=<name>` plus `filter`); no request while
* `evalName` is empty. `data` is only ever rows of the current `evalName`:
Expand Down
15 changes: 8 additions & 7 deletions apps/agent-forge/src/screens/monitoring/MetricRow.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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 };
Expand Down
10 changes: 6 additions & 4 deletions apps/agent-forge/src/screens/monitoring/MonitoringScreen.tsx
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -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,
);
Expand All @@ -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(() => {
Expand Down Expand Up @@ -135,7 +137,7 @@ export function MonitoringScreen() {
<div className="mon-screen">
{header}
<ObserveError error={error} />
<MetricRow rows={rows} from={chart.from} win={win} loading={chartsLoading} />
<MetricRow rows={rows} chart={chart} win={win} loading={chartsLoading} />
<section className="mon-fleet" aria-label="Fleet">
<Card as="div" title="Agents" description="Registered agents: today's runs and error rate, last-24h p95." padding="none">
<FleetTable
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,6 @@ describe('MonitoringScreen', () => {

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),
Expand All @@ -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(<MonitoringScreen />);
await waitFor(() => drawn(c, 'Runs'));
Expand All @@ -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(<MonitoringScreen />);
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(<MonitoringScreen />);
await waitFor(() => drawn(c, 'Runs'));
let release: () => void = () => undefined;
const stalled = new Promise<void>((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)]);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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);
});
});

Expand Down Expand Up @@ -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);
Expand All @@ -78,27 +135,40 @@ 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);
expect(m.errorRate.delta).toBeCloseTo(0); // 10% vs 10%
});

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();
Expand Down
Loading
Loading