diff --git a/packages/app/cypress/e2e/collectivex.cy.ts b/packages/app/cypress/e2e/collectivex.cy.ts index a08d48a19..ebdc623f7 100644 --- a/packages/app/cypress/e2e/collectivex.cy.ts +++ b/packages/app/cypress/e2e/collectivex.cy.ts @@ -28,6 +28,21 @@ const incompleteDataset = buildDataset({ conclusion: 'failure', }, }); +const kvDataset = buildDataset({ + shards: [makeRawShard()], + kv: [ + {}, + { + sku: 'mi355x', + backend: 'mori-io', + fabric: 'rdma', + vendor: 'amd', + status: 'invalid', + reasons: ['transfer verification failed'], + }, + ], + meta: { run_id: '162', generated_at: '2026-08-07T12:20:00Z', source_sha: 'e'.repeat(40) }, +}); const ADMIN_TOKEN_KEY = 'collectivex-admin-token'; function installRuns(bodies: CollectiveXDataset[] = [dataset]) { @@ -477,3 +492,43 @@ describe('CollectiveX availability states', () => { cy.then(() => expect(availabilityRequests).to.eq(0)); }); }); + +describe('CollectiveX kv-transfer card', () => { + it('renders kv cases with bandwidth-bound cells and per-case outcomes', () => { + installRuns([kvDataset]); + installRun(kvDataset); + openCollectiveX(); + cy.get('[data-testid="collectivex-kv-table"]') + .should('be.visible') + .and('contain.text', 'KV-cache transfer') + .and('contain.text', '2 cases') + .and('contain.text', '1 measured'); + cy.get('[data-testid="collectivex-kv-table-table"]').within(() => { + // The measured gb200 nixl case: bulk ceiling, paged-64 at batch 1 and + // at the largest measured batch, paged-16, and the handoff latency. + cy.contains('td', 'GB200').parent().as('measured'); + cy.get('@measured').should('contain.text', 'nixl').and('contain.text', 'kv-dsv4'); + cy.get('@measured').should('contain.text', '89.41'); + cy.get('@measured').should('contain.text', '7.39'); + cy.get('@measured').should('contain.text', '15.12 (b16)'); + cy.get('@measured').should('contain.text', '2.72'); + cy.get('@measured').should('contain.text', '24.8'); + // The failed mori-io case keeps its outcome and reason, with no cells. + cy.contains('td', 'MI355X').parent().as('failed'); + cy.get('@failed').should('contain.text', 'mori-io').and('contain.text', 'invalid'); + cy.get('@failed').should('contain.text', 'transfer-verification-failed'); + }); + // KV cases count into the header stats alongside EP cases. + cy.get('[data-testid="collectivex-display"]').should( + 'contain.text', + `${kvDataset.run.measured_cases}/${kvDataset.run.requested_cases}`, + ); + }); + + it('renders no kv card for an EP-only run', () => { + installRuns(); + installRun(); + openCollectiveX(); + cy.get('[data-testid="collectivex-kv-table"]').should('not.exist'); + }); +}); diff --git a/packages/app/src/components/collectivex/CollectiveXDisplay.tsx b/packages/app/src/components/collectivex/CollectiveXDisplay.tsx index 1b082039d..6c3c1c519 100644 --- a/packages/app/src/components/collectivex/CollectiveXDisplay.tsx +++ b/packages/app/src/components/collectivex/CollectiveXDisplay.tsx @@ -26,6 +26,7 @@ import { useLocale } from '@/lib/use-locale'; import { CollectiveXChart } from './CollectiveXChart'; import { CollectiveXInventory } from './CollectiveXInventory'; +import { CollectiveXKvTable } from './CollectiveXKvTable'; import { CollectiveXRunsTable } from './CollectiveXRunsTable'; import { collectiveXColorKey, @@ -1053,6 +1054,7 @@ export default function CollectiveXDisplay() { /> + `${dataset.run.run_id}:${dataset.run.run_attempt}`).join(',')}`} datasets={datasets} diff --git a/packages/app/src/components/collectivex/CollectiveXInventory.tsx b/packages/app/src/components/collectivex/CollectiveXInventory.tsx index c3e9ec57a..11c25d671 100644 --- a/packages/app/src/components/collectivex/CollectiveXInventory.tsx +++ b/packages/app/src/components/collectivex/CollectiveXInventory.tsx @@ -143,11 +143,10 @@ export function CollectiveXInventory({ datasets }: { datasets: CollectiveXDatase const points = rows.flatMap((item) => item.points); const measured = points.filter((point) => point.terminal_status === 'measured').length; const unsupported = points.filter((point) => point.terminal_status === 'unsupported').length; - const measuredCases = datasets.reduce((sum, dataset) => sum + dataset.run.measured_cases, 0); - const unsupportedCases = datasets.reduce( - (sum, dataset) => sum + dataset.run.unsupported_cases, - 0, - ); + // Counted from this table's own EP coverage rows: the run-level totals also + // include kv-transfer cases, which live in their own card, not here. + const measuredCases = rows.filter((row) => row.outcome === 'success').length; + const unsupportedCases = rows.filter((row) => row.outcome === 'unsupported').length; const terminalPoints = datasets.reduce((sum, dataset) => sum + dataset.run.terminal_points, 0); const requestedPoints = datasets.reduce((sum, dataset) => sum + dataset.run.requested_points, 0); diff --git a/packages/app/src/components/collectivex/CollectiveXKvTable.tsx b/packages/app/src/components/collectivex/CollectiveXKvTable.tsx new file mode 100644 index 000000000..88a23c12b --- /dev/null +++ b/packages/app/src/components/collectivex/CollectiveXKvTable.tsx @@ -0,0 +1,153 @@ +'use client'; + +import { useMemo } from 'react'; + +import { Badge } from '@/components/ui/badge'; +import { Card } from '@/components/ui/card'; +import { type DataTableColumn, DataTable } from '@/components/ui/data-table'; +import { useLocale } from '@/lib/use-locale'; + +import { collectiveXKvCell } from './data'; +import type { CollectiveXDataset, CollectiveXKvCase, CollectiveXOutcome } from './types'; + +type CollectiveXRunKvCase = CollectiveXKvCase & { run_id: string }; + +const STRINGS = { + en: { + heading: 'KV-cache transfer', + description: + 'Prefill-to-decode KV handoff (2 nodes x 1 GPU, DeepSeek-V4-Pro cache as vLLM allocates it). ' + + 'Paged rows move per-request layer-major descriptor lists over randomized block tables; ' + + 'bulk is the single-descriptor wire ceiling. GB/s is burst-aggregate pull at the largest ISL; ' + + 'b1/bmax are requests posted per burst.', + }, + zh: { + heading: 'KV 缓存传输', + description: + '预填充到解码的 KV 交接(2 节点 x 1 GPU,按 vLLM 为 DeepSeek-V4-Pro 分配的缓存布局)。' + + '分页行按随机块表以逐层描述符列表搬运每个请求;bulk 为单描述符线速上限。' + + 'GB/s 为最大 ISL 处按突发聚合的 pull 带宽;b1/bmax 表示每次突发提交的请求数。', + }, +} as const; + +const OUTCOME_CLASS: Record = { + success: 'border-emerald-600/40 bg-emerald-500/10 text-emerald-700 dark:text-emerald-300', + unsupported: 'border-zinc-500/40 bg-zinc-500/10 text-zinc-700 dark:text-zinc-300', + failed: 'border-red-700/50 bg-red-700/10 text-red-800 dark:text-red-300', + invalid: 'border-red-600/40 bg-red-500/10 text-red-700 dark:text-red-300', + diagnostic: 'border-amber-600/40 bg-amber-500/10 text-amber-700 dark:text-amber-300', + pending: 'border-zinc-500/40 bg-zinc-500/5 text-muted-foreground', +}; + +function formatGbps(value: number | null | undefined): string { + return value === null || value === undefined ? '-' : value.toFixed(value >= 100 ? 0 : 2); +} + +function cellsOf(row: CollectiveXRunKvCase) { + return { + p64b1: collectiveXKvCell(row.rows, 'paged', 64, 'min'), + p64bmax: collectiveXKvCell(row.rows, 'paged', 64, 'max'), + p16b1: collectiveXKvCell(row.rows, 'paged', 16, 'min'), + bulk: collectiveXKvCell(row.rows, 'bulk', null, 'min'), + }; +} + +export function CollectiveXKvTable({ datasets }: { datasets: CollectiveXDataset[] }) { + const locale = useLocale(); + const strings = STRINGS[locale === 'zh' ? 'zh' : 'en']; + const rows = useMemo( + () => + datasets.flatMap((dataset) => + (dataset.kv ?? []).map((item) => ({ ...item, run_id: dataset.run.run_id })), + ), + [datasets], + ); + const columns = useMemo[]>( + () => [ + { + header: 'Run', + cell: (row) => #{row.run_id}, + sortValue: (row) => Number(row.run_id), + className: 'whitespace-nowrap', + }, + { header: 'SKU', cell: (row) => row.sku.toUpperCase(), sortValue: (row) => row.sku }, + { + header: 'Backend', + cell: (row) => row.backend, + sortValue: (row) => row.backend, + className: 'whitespace-nowrap', + }, + { header: 'Fabric', cell: (row) => row.fabric, sortValue: (row) => row.fabric }, + { header: 'Workload', cell: (row) => row.workload, sortValue: (row) => row.workload }, + { header: 'Precision', cell: (row) => row.precision, sortValue: (row) => row.precision }, + { + header: 'Outcome', + cell: (row) => ( +
+ + {row.outcome} + + {(row.detail || row.reason) && ( +

{row.detail ?? row.reason}

+ )} +
+ ), + sortValue: (row) => `${row.outcome} ${row.reason ?? ''}`, + }, + { + header: 'Bulk GB/s', + cell: (row) => formatGbps(cellsOf(row).bulk?.gbps_p50), + sortValue: (row) => cellsOf(row).bulk?.gbps_p50 ?? -1, + className: 'text-right tabular-nums', + }, + { + header: 'p64 GB/s b1', + cell: (row) => formatGbps(cellsOf(row).p64b1?.gbps_p50), + sortValue: (row) => cellsOf(row).p64b1?.gbps_p50 ?? -1, + className: 'text-right tabular-nums', + }, + { + header: 'p64 GB/s bmax', + cell: (row) => { + const cell = cellsOf(row).p64bmax; + if (!cell) return '-'; + return `${formatGbps(cell.gbps_p50)} (b${cell.batch})`; + }, + sortValue: (row) => cellsOf(row).p64bmax?.gbps_p50 ?? -1, + className: 'text-right tabular-nums whitespace-nowrap', + }, + { + header: 'p16 GB/s b1', + cell: (row) => formatGbps(cellsOf(row).p16b1?.gbps_p50), + sortValue: (row) => cellsOf(row).p16b1?.gbps_p50 ?? -1, + className: 'text-right tabular-nums', + }, + { + header: 'Handoff ms', + cell: (row) => { + const cell = cellsOf(row).p64b1; + return cell ? cell.latency_ms.p50.toFixed(1) : '-'; + }, + sortValue: (row) => cellsOf(row).p64b1?.latency_ms.p50 ?? -1, + className: 'text-right tabular-nums', + }, + ], + [], + ); + if (rows.length === 0) return null; + const measured = rows.filter((row) => row.outcome === 'success').length; + return ( + +

{strings.heading}

+

+ {rows.length} cases · {measured} measured · {strings.description} +

+ +
+ ); +} diff --git a/packages/app/src/components/collectivex/data.test.ts b/packages/app/src/components/collectivex/data.test.ts index 3ee8ab36c..4777bdee3 100644 --- a/packages/app/src/components/collectivex/data.test.ts +++ b/packages/app/src/components/collectivex/data.test.ts @@ -12,8 +12,9 @@ import { metricValue, seriesMatchesSelection, type CollectiveXSeriesSelection, + collectiveXKvCell, } from './data'; -import type { CollectiveXPercentiles, CollectiveXSeries } from './types'; +import type { CollectiveXKvRow, CollectiveXPercentiles, CollectiveXSeries } from './types'; import { makeCollectiveXDataset, makeCollectiveXSeries } from './test-fixture'; const dataset = makeCollectiveXDataset(); @@ -280,3 +281,41 @@ describe('chartPoints', () => { expect(points).toHaveLength(scaleUp.points.length); }); }); + +const kvRow = (overrides: Partial): CollectiveXKvRow => ({ + kind: 'paged', + isl: 32768, + page_tokens: 64, + batch: 1, + op: 'pull', + descs: 20302, + req_bytes: 183000000, + prep_ms: 1.2, + latency_ms: { p50: 24.8, p95: 26, min: 24.1, max: 26.4, n: 24 }, + gbps_p50: 7.39, + verify_passed: true, + ...overrides, +}); + +describe('collectiveXKvCell', () => { + it('picks the largest-ISL pull row at the requested batch extreme', () => { + const rows = [ + kvRow({ isl: 4096, gbps_p50: 5 }), + kvRow({ gbps_p50: 7.39 }), + kvRow({ batch: 16, gbps_p50: 15.12 }), + kvRow({ op: 'push', batch: 16, gbps_p50: 99 }), + ]; + expect(collectiveXKvCell(rows, 'paged', 64, 'min')?.gbps_p50).toBe(7.39); + expect(collectiveXKvCell(rows, 'paged', 64, 'max')?.gbps_p50).toBe(15.12); + }); + + it('returns null for an unmeasured family', () => { + expect(collectiveXKvCell([kvRow({})], 'paged', 16, 'min')).toBeNull(); + expect(collectiveXKvCell([], 'bulk', null, 'min')).toBeNull(); + }); + + it('selects bulk rows by their null page size', () => { + const rows = [kvRow({}), kvRow({ kind: 'bulk', page_tokens: null, gbps_p50: 89.41 })]; + expect(collectiveXKvCell(rows, 'bulk', null, 'min')?.gbps_p50).toBe(89.41); + }); +}); diff --git a/packages/app/src/components/collectivex/data.ts b/packages/app/src/components/collectivex/data.ts index 4cc31f478..53512a535 100644 --- a/packages/app/src/components/collectivex/data.ts +++ b/packages/app/src/components/collectivex/data.ts @@ -1,6 +1,7 @@ import type { CollectiveXChartPoint, CollectiveXComponent, + CollectiveXKvRow, CollectiveXMode, CollectiveXOperation, CollectiveXPercentile, @@ -198,3 +199,26 @@ export function chartPoints( }), ); } + +/** + * The kv table cell selector, mirroring the harness's summarize: the + * largest-ISL pull row of a (kind, page) family — the bandwidth-bound point — + * at its smallest or largest measured batch. Null when the family was not + * measured (e.g. a page size the sweep dropped). + */ +export function collectiveXKvCell( + rows: CollectiveXKvRow[], + kind: CollectiveXKvRow['kind'], + pageTokens: number | null, + batch: 'min' | 'max', +): CollectiveXKvRow | null { + const matching = rows.filter( + (row) => row.kind === kind && row.page_tokens === pageTokens && row.op === 'pull', + ); + if (matching.length === 0) return null; + const isl = Math.max(...matching.map((row) => row.isl)); + const atIsl = matching.filter((row) => row.isl === isl); + const pick = (better: (a: number, b: number) => boolean) => + atIsl.reduce((best, row) => (better(row.batch, best.batch) ? row : best)); + return batch === 'min' ? pick((a, b) => a < b) : pick((a, b) => a > b); +} diff --git a/packages/app/src/components/collectivex/types.ts b/packages/app/src/components/collectivex/types.ts index 17b65a2a1..b12617463 100644 --- a/packages/app/src/components/collectivex/types.ts +++ b/packages/app/src/components/collectivex/types.ts @@ -19,6 +19,9 @@ export type { CollectiveXCoverage, CollectiveXCoveragePoint, CollectiveXDataset, + CollectiveXKvCase, + CollectiveXKvLatency, + CollectiveXKvRow, CollectiveXMode, CollectiveXOperation, CollectiveXOutcome, diff --git a/packages/db/src/collectivex/reader.test.ts b/packages/db/src/collectivex/reader.test.ts index 91e800e37..7f38b257c 100644 --- a/packages/db/src/collectivex/reader.test.ts +++ b/packages/db/src/collectivex/reader.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest'; -import { buildDatasetFromNeutral } from './reader'; +import { buildDatasetFromNeutral, buildRunSummary } from './reader'; import { buildDataset, makeCollectiveXDataset, @@ -233,3 +233,80 @@ describe('CollectiveX artifact assembly', () => { ).toThrow(/version/); }); }); + +describe('CollectiveX kv-transfer assembly', () => { + it('assembles kv cases beside EP coverage without cross-contamination', () => { + const dataset = buildDataset({ shards: [makeRawShard()], kv: [{}] }); + expect(dataset.coverage).toHaveLength(1); + expect(dataset.series).toHaveLength(1); + expect(dataset.kv).toHaveLength(1); + const kase = dataset.kv![0]; + expect(kase).toMatchObject({ + sku: 'gb200', + backend: 'nixl', + fabric: 'rdma', + workload: 'kv-dsv4', + precision: 'fp8', + outcome: 'success', + vendor: 'nvidia', + }); + expect(kase.rows).toHaveLength(4); + expect(kase.rows[0]).toMatchObject({ + kind: 'paged', + isl: 32768, + page_tokens: 64, + batch: 1, + op: 'pull', + gbps_p50: 7.39, + verify_passed: true, + }); + // KV cases count into the run totals (the visibility gate) but not points. + expect(dataset.run).toMatchObject({ + requested_cases: 2, + measured_cases: 2, + kv_requested_cases: 1, + kv_measured_cases: 1, + }); + expect(dataset.run.covered_skus).toContain('gb200'); + expect(dataset.run.requested_points).toBe(10); + }); + + it('coerces the kv entrypoint string version against the numeric matrix version', () => { + const dataset = buildDataset({ kv: [{ version: '1' }] }); + expect(dataset.kv![0].outcome).toBe('success'); + }); + + it('carries a failed kv shard outcome and reason without rows', () => { + const dataset = buildDataset({ + kv: [{ status: 'invalid', reasons: ['transfer verification failed'] }], + }); + const kase = dataset.kv![0]; + expect(kase.outcome).toBe('invalid'); + expect(kase.reason).toBe('transfer-verification-failed'); + expect(kase.rows).toHaveLength(0); + expect(dataset.run.failed_cases).toBe(1); + }); + + it('marks a requested kv case with no shard as pending', () => { + const dataset = buildDataset({ kv: [{ omitShard: true }] }); + expect(dataset.kv![0]).toMatchObject({ outcome: 'pending', vendor: null, rows: [] }); + expect(dataset.run.terminal_cases).toBe(1); // the EP shard only + }); + + it('summarizes kv case counts for the run picker', () => { + const dataset = buildDataset({ + kv: [{}, { sku: 'mi355x', backend: 'mori-io', vendor: 'amd' }], + }); + const summary = buildRunSummary(dataset); + expect(summary.kv_cases).toEqual({ requested: 2, measured: 2 }); + expect(summary.requested_cases).toBe(3); + }); + + it('keeps a kv-only run visible through the case totals', () => { + const kv = buildDataset({ shards: [], kv: [{}] }); + expect(kv.coverage).toHaveLength(0); + expect(kv.series).toHaveLength(0); + expect(kv.run.requested_cases).toBe(1); + expect(kv.run.covered_skus).toEqual(['gb200']); + }); +}); diff --git a/packages/db/src/collectivex/reader.ts b/packages/db/src/collectivex/reader.ts index 55db606f1..b3f4fb6cc 100644 --- a/packages/db/src/collectivex/reader.ts +++ b/packages/db/src/collectivex/reader.ts @@ -15,6 +15,9 @@ import type { CollectiveXCoverage, CollectiveXCoveragePoint, CollectiveXDataset, + CollectiveXKvCase, + CollectiveXKvLatency, + CollectiveXKvRow, CollectiveXMode, CollectiveXOutcome, CollectiveXPercentiles, @@ -39,6 +42,10 @@ interface RawCase { scale_up_domain: number; scale_up_transport: string; scale_out_transport: string | null; + /** `kv-transfer` on KV handoff cases; absent on EP cases. */ + suite?: string; + /** KV workload preset name (e.g. `kv-dsv4`); absent on EP cases. */ + workload?: string; } interface RawComponent { @@ -54,8 +61,25 @@ interface RawRow { byte_provenance: Record; } +// KV shards report per-burst rows instead of per-ladder-token rows; the two +// families share the shard envelope and are told apart by `case.suite`. +interface RawKvRow { + kind: string; + isl: number; + page_tokens: number | null; + batch?: number; + op: string; + descs: number; + req_bytes: number; + prep_ms?: number; + latency_ms: CollectiveXKvLatency; + gbps_p50: number; + verify?: { passed: boolean; detail?: string }; +} + interface RawShard { - version: number; + /** Numeric in EP artifacts; the kv entrypoint emitted it as a string. */ + version: number | string; record_type: 'case-attempt'; identity: { case_id: string; @@ -279,6 +303,81 @@ function terminalPoints( })); } +function isKvCase(kase: RawCase): boolean { + return kase.suite === 'kv-transfer'; +} + +function mapKvRow(row: RawKvRow): CollectiveXKvRow { + return { + kind: row.kind === 'bulk' ? 'bulk' : 'paged', + isl: row.isl, + page_tokens: row.page_tokens ?? null, + // Rows predating the batch dimension measured one request per burst. + batch: row.batch ?? 1, + op: row.op === 'push' ? 'push' : 'pull', + descs: row.descs, + req_bytes: row.req_bytes, + prep_ms: row.prep_ms ?? 0, + latency_ms: row.latency_ms, + gbps_p50: row.gbps_p50, + verify_passed: row.verify?.passed ?? true, + }; +} + +function buildKvCases( + requestedCases: RawMatrix['requested_cases'], + successful: Map, + terminal: Map, + hiddenCaseIds: Set, +): CollectiveXKvCase[] { + return requestedCases.flatMap((requested) => { + const kase = requested.case; + const caseId = kase.case_id; + if (!isKvCase(kase) || !caseId || hiddenCaseIds.has(caseId)) return []; + const measured = successful.get(caseId); + const failed = terminal.get(caseId); + let outcome: CollectiveXOutcome; + let reason: string | null; + if (measured) { + outcome = 'success'; + reason = null; + } else if (failed) { + outcome = toOutcome(failed.shard.outcome.status); + reason = reasonId(failed.shard.outcome.reasons?.[0] ?? outcome); + } else if (requested.disposition === 'unsupported') { + outcome = 'unsupported'; + reason = reasonId(requested.reason ?? outcome); + } else { + outcome = 'pending'; + reason = 'pending'; + } + const fabric = kase.mode ?? 'rdma'; + const workload = kase.workload ?? 'kv'; + const precision = toPrecision(kase.precision); + const shard = measured ?? failed; + return [ + { + case_id: caseId, + label: `${requested.sku} · ${kase.backend} · ${fabric} · ${workload} · ${precision}`, + disposition: requested.disposition, + sku: requested.sku, + vendor: shard?.vendor ?? null, + backend: kase.backend, + fabric, + workload, + precision, + topology: topologyOf(kase), + outcome, + reason, + detail: requested.detail ?? null, + rows: measured + ? (measured.shard.measurement.rows as unknown as RawKvRow[]).map(mapKvRow) + : [], + }, + ]; + }); +} + export function buildDatasetFromNeutral( matrixRaw: unknown, docs: unknown[], @@ -288,7 +387,7 @@ export function buildDatasetFromNeutral( const shards = docs.flatMap((doc) => { const shard = shardOf(doc); if (!shard) return []; - if (shard.version !== matrix.version) throw new Error('CollectiveX version mismatch'); + if (Number(shard.version) !== matrix.version) throw new Error('CollectiveX version mismatch'); return [shard]; }); const supportedShards = shards.flatMap((shard): SupportedShard[] => { @@ -315,7 +414,7 @@ export function buildDatasetFromNeutral( const coverage: CollectiveXCoverage[] = matrix.requested_cases.flatMap((requested) => { const kase = requested.case; const caseId = kase.case_id; - if (!caseId || hiddenCaseIds.has(caseId)) return []; + if (isKvCase(kase) || !caseId || hiddenCaseIds.has(caseId)) return []; const measured = successful.get(caseId)?.shard; const failed = terminal.get(caseId)?.shard; let outcome: CollectiveXOutcome; @@ -358,27 +457,43 @@ export function buildDatasetFromNeutral( }, ]; }); + const kv = buildKvCases(matrix.requested_cases, successful, terminal, hiddenCaseIds); const points = coverage.flatMap((item) => item.points); + // KV cases count into the run's case totals (the run picker's visibility + // gate is `requested_cases > 0`, and a kv-only sweep is a real run), but + // carry no ladder points — point totals stay EP-only. return { version: matrix.version, run: { ...run, - requested_cases: coverage.length, - terminal_cases: coverage.filter((item) => - item.points.every((point) => point.terminal_status !== 'pending'), - ).length, - measured_cases: coverage.filter((item) => item.outcome === 'success').length, - unsupported_cases: coverage.filter((item) => item.outcome === 'unsupported').length, - failed_cases: coverage.filter((item) => - ['failed', 'invalid', 'diagnostic'].includes(item.outcome), - ).length, + requested_cases: coverage.length + kv.length, + terminal_cases: + coverage.filter((item) => item.points.every((point) => point.terminal_status !== 'pending')) + .length + kv.filter((item) => item.outcome !== 'pending').length, + measured_cases: + coverage.filter((item) => item.outcome === 'success').length + + kv.filter((item) => item.outcome === 'success').length, + unsupported_cases: + coverage.filter((item) => item.outcome === 'unsupported').length + + kv.filter((item) => item.outcome === 'unsupported').length, + failed_cases: + coverage.filter((item) => ['failed', 'invalid', 'diagnostic'].includes(item.outcome)) + .length + + kv.filter((item) => ['failed', 'invalid', 'diagnostic'].includes(item.outcome)).length, requested_points: points.length, terminal_points: points.filter((point) => point.terminal_status !== 'pending').length, measured_points: points.filter((point) => point.terminal_status === 'measured').length, - covered_skus: [...new Set(coverage.map((item) => item.sku))].toSorted(), + covered_skus: [ + ...new Set([...coverage.map((item) => item.sku), ...kv.map((item) => item.sku)]), + ].toSorted(), + kv_requested_cases: kv.length, + kv_measured_cases: kv.filter((item) => item.outcome === 'success').length, }, coverage, - series: [...successful.values()].map(buildSeries), + series: [...successful.values()] + .filter(({ shard }) => !isKvCase(shard.identity.case_factors.case)) + .map(buildSeries), + kv, }; } @@ -399,6 +514,10 @@ export function buildRunSummary(dataset: CollectiveXDataset): CollectiveXRunSumm unsupported: run.unsupported_cases, failed: run.failed_cases, }, + kv_cases: { + requested: run.kv_requested_cases ?? 0, + measured: run.kv_measured_cases ?? 0, + }, }; } diff --git a/packages/db/src/collectivex/test-fixture.ts b/packages/db/src/collectivex/test-fixture.ts index ccb32f6e8..39a121743 100644 --- a/packages/db/src/collectivex/test-fixture.ts +++ b/packages/db/src/collectivex/test-fixture.ts @@ -177,6 +177,130 @@ export function makeRawMatrix(requested: RequestedCaseSpec[], version = 1): Json }; } +export interface KvOverrides { + sku?: string; + backend?: string; + fabric?: string; + workload?: string; + precision?: string; + vendor?: string; + status?: string; + reasons?: string[]; + disposition?: 'runnable' | 'unsupported'; + reason?: string; + /** String on purpose by default: the kv entrypoint emitted `version: '1'`. */ + version?: number | string; + rows?: Partial[]; + omitShard?: boolean; +} + +interface KvRowSpec { + kind: string; + isl: number; + page_tokens: number | null; + batch: number; + op: string; + gbps_p50: number; + latency_p50: number; + verify_passed: boolean; +} + +function makeRawKvRow(spec: Partial): Json { + const latency = spec.latency_p50 ?? 24.77; + return { + kind: spec.kind ?? 'paged', + preset: 'dsv4', + isl: spec.isl ?? 32768, + page_tokens: spec.page_tokens === undefined ? 64 : spec.page_tokens, + layers: 61, + page_bytes: spec.kind === 'bulk' ? null : 9216, + descs: spec.kind === 'bulk' ? 1 : 20302, + req_bytes: 183000000, + batch: spec.batch ?? 1, + op: spec.op ?? 'pull', + prep_ms: 1.2, + latency_ms: { + p50: latency, + p95: latency * 1.05, + min: latency * 0.98, + max: latency * 1.1, + n: 24, + }, + gbps_p50: spec.gbps_p50 ?? 7.39, + verify: { passed: spec.verify_passed ?? true, detail: '' }, + }; +} + +function kvCaseIdOf(options: KvOverrides): string { + return `${options.sku ?? 'gb200'}-${options.backend ?? 'nixl'}-${options.workload ?? 'kv-dsv4'}-${options.fabric ?? 'rdma'}-xfer-ep2-paged-${options.precision ?? 'fp8'}`; +} + +/** + * A kv-transfer shard + its matrix requested-case entry. Mirrors the real + * artifacts: the shard's case carries only the identity factors while the + * matrix entry carries the full case (isl_ladder, batch_sizes, topology). + */ +export function makeKvFixture(options: KvOverrides = {}): { + shard: Json | null; + requested: RequestedCaseSpec; +} { + const caseId = kvCaseIdOf(options); + const sku = options.sku ?? 'gb200'; + const identityCase: Json = { + backend: options.backend ?? 'nixl', + workload: options.workload ?? 'kv-dsv4', + mode: options.fabric ?? 'rdma', + phase: 'xfer', + ep: 2, + routing: 'paged', + precision: options.precision ?? 'fp8', + suite: 'kv-transfer', + }; + const rows = options.rows ?? [ + { kind: 'paged', page_tokens: 64, batch: 1 }, + { kind: 'paged', page_tokens: 64, batch: 16, gbps_p50: 15.12, latency_p50: 193.7 }, + { kind: 'paged', page_tokens: 16, batch: 1, gbps_p50: 2.72, latency_p50: 67.3 }, + { kind: 'bulk', page_tokens: null, batch: 1, gbps_p50: 89.41, latency_p50: 2.05 }, + ]; + const shard: Json | null = options.omitShard + ? null + : { + version: options.version ?? '1', + record_type: 'case-attempt', + identity: { case_id: caseId, case_factors: { sku, case: identityCase } }, + implementation: { name: options.backend ?? 'nixl' }, + runtime: { vendor: options.vendor ?? 'nvidia' }, + measurement: { rows: rows.map(makeRawKvRow) }, + outcome: { + status: options.status ?? 'success', + ...(options.reasons ? { reasons: options.reasons } : {}), + }, + }; + const requested: RequestedCaseSpec = { + caseId, + sku, + disposition: options.disposition, + reason: options.reason, + case: { + ...identityCase, + case_id: caseId, + isl_ladder: '512 4096 32768', + page_tokens: '16 64', + batch_sizes: '1 4 16', + ops: 'pull push', + nodes: 2, + gpus_per_node: 1, + scale_up_domain: 72, + scope: 'scale-out', + scale_up_transport: 'mnnvl', + scale_out_transport: options.fabric ?? 'rdma', + transport: options.fabric ?? 'rdma', + topology_class: `${sku}-kv-${options.fabric ?? 'rdma'}`, + }, + }; + return { shard, requested }; +} + export function makeRunMeta( overrides: Partial = {}, ): CollectiveXNeutralRunMeta { @@ -194,12 +318,19 @@ export function buildDataset( options: { shards?: Json[]; requestedCases?: RequestedCaseSpec[]; + kv?: KvOverrides[]; meta?: Partial; } = {}, ): CollectiveXDataset { const shards = options.shards ?? [makeRawShard()]; - const requested = [...shards.map(requestedFromShard), ...(options.requestedCases ?? [])]; - return buildDatasetFromNeutral(makeRawMatrix(requested), shards, makeRunMeta(options.meta)); + const kvFixtures = (options.kv ?? []).map(makeKvFixture); + const requested = [ + ...shards.map(requestedFromShard), + ...kvFixtures.map((fixture) => fixture.requested), + ...(options.requestedCases ?? []), + ]; + const docs = [...shards, ...kvFixtures.flatMap((fixture) => fixture.shard ?? [])]; + return buildDatasetFromNeutral(makeRawMatrix(requested), docs, makeRunMeta(options.meta)); } export function makeCollectiveXSeries(overrides: ShardOverrides = {}): CollectiveXSeries { diff --git a/packages/db/src/collectivex/types.ts b/packages/db/src/collectivex/types.ts index 9bc2d2e3c..450fb51dc 100644 --- a/packages/db/src/collectivex/types.ts +++ b/packages/db/src/collectivex/types.ts @@ -108,6 +108,52 @@ export interface CollectiveXCoverage { detail: string | null; } +/** kv-transfer latency percentiles (the suite reports ms, not us). */ +export interface CollectiveXKvLatency { + p50: number; + p95: number; + min: number; + max: number; + n: number; +} + +/** One measured kv-transfer grid point (a burst of `batch` requests). */ +export interface CollectiveXKvRow { + kind: 'paged' | 'bulk'; + isl: number; + page_tokens: number | null; + batch: number; + op: 'pull' | 'push'; + descs: number; + req_bytes: number; + prep_ms: number; + latency_ms: CollectiveXKvLatency; + gbps_p50: number; + verify_passed: boolean; +} + +/** + * One kv-transfer matrix case (2 nodes x 1 GPU: the per-worker prefill/decode + * pair). Fabric is the case's declared lane (`rdma` | `mnnvl`); rows are empty + * unless a successful shard measured the case. + */ +export interface CollectiveXKvCase { + case_id: string; + label: string; + disposition: 'runnable' | 'unsupported'; + sku: string; + vendor: 'nvidia' | 'amd' | null; + backend: string; + fabric: string; + workload: string; + precision: CollectiveXPrecision; + topology: CollectiveXTopology; + outcome: CollectiveXOutcome; + reason: string | null; + detail: string | null; + rows: CollectiveXKvRow[]; +} + export interface CollectiveXRun { run_id: string; run_attempt: number; @@ -123,6 +169,9 @@ export interface CollectiveXRun { terminal_points: number; measured_points: number; covered_skus: string[]; + /** kv-transfer case counts; absent on datasets read before the kv suite. */ + kv_requested_cases?: number; + kv_measured_cases?: number; } export interface CollectiveXDataset { @@ -130,6 +179,8 @@ export interface CollectiveXDataset { run: CollectiveXRun; coverage: CollectiveXCoverage[]; series: CollectiveXSeries[]; + /** kv-transfer cases; absent on datasets captured before the kv suite. */ + kv?: CollectiveXKvCase[]; } export interface CollectiveXRunSummary { @@ -143,4 +194,6 @@ export interface CollectiveXRunSummary { requested_points: number; terminal_points: number; terminal_counts: { measured: number; unsupported: number; failed: number }; + /** kv-transfer case counts; absent on summaries stored before the kv suite. */ + kv_cases?: { requested: number; measured: number }; }