From 926437992c0e207b7e1f9192e7fa0297f36215bf Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sat, 17 Oct 2026 14:34:06 -0700 Subject: [PATCH 1/3] feat(biometrics): export frozen calibration datasets and fit profiles on a laptop, since a 4-night backtest takes an hour on Pod 5 --- .../datasets/[id]/export/route.ts | 87 +++++++++ docs/biometrics-datasets.md | 15 ++ package.json | 1 + scripts/biometrics-local-fit/README.md | 89 +++++++++ scripts/biometrics-local-fit/bundle.test.ts | 105 +++++++++++ scripts/biometrics-local-fit/bundle.ts | 110 +++++++++++ scripts/biometrics-local-fit/cli.ts | 106 +++++++++++ scripts/biometrics-local-fit/fit.ts | 172 ++++++++++++++++++ scripts/biometrics-local-fit/grid.ts | 40 ++++ scripts/biometrics-local-fit/local_replay.py | 34 ++++ scripts/biometrics-local-fit/report.test.ts | 118 ++++++++++++ scripts/biometrics-local-fit/report.ts | 120 ++++++++++++ src/biometrics/backtest/bundle-tar.ts | 59 ++++++ src/biometrics/backtest/coordinator-lanes.ts | 2 +- src/biometrics/backtest/coordinator-tasks.ts | 4 + src/biometrics/backtest/dataset-export.ts | 147 +++++++++++++++ .../backtest/tests/dataset-export.test.ts | 146 +++++++++++++++ .../backtest/tests/fixtures/export.ts | 63 +++++++ src/biometrics/backtest/waveform-child.ts | 7 +- src/mcp/README.md | 1 + src/mcp/tests/server.test.ts | 9 +- src/mcp/tools/biometrics.ts | 7 + src/server/routers/biometricTuning.ts | 8 + 23 files changed, 1446 insertions(+), 4 deletions(-) create mode 100644 app/api/biometric-tuning/datasets/[id]/export/route.ts create mode 100644 scripts/biometrics-local-fit/README.md create mode 100644 scripts/biometrics-local-fit/bundle.test.ts create mode 100644 scripts/biometrics-local-fit/bundle.ts create mode 100644 scripts/biometrics-local-fit/cli.ts create mode 100644 scripts/biometrics-local-fit/fit.ts create mode 100644 scripts/biometrics-local-fit/grid.ts create mode 100644 scripts/biometrics-local-fit/local_replay.py create mode 100644 scripts/biometrics-local-fit/report.test.ts create mode 100644 scripts/biometrics-local-fit/report.ts create mode 100644 src/biometrics/backtest/bundle-tar.ts create mode 100644 src/biometrics/backtest/dataset-export.ts create mode 100644 src/biometrics/backtest/tests/dataset-export.test.ts create mode 100644 src/biometrics/backtest/tests/fixtures/export.ts diff --git a/app/api/biometric-tuning/datasets/[id]/export/route.ts b/app/api/biometric-tuning/datasets/[id]/export/route.ts new file mode 100644 index 00000000..ed659e3e --- /dev/null +++ b/app/api/biometric-tuning/datasets/[id]/export/route.ts @@ -0,0 +1,87 @@ +import { NextResponse } from 'next/server' +import { z } from 'zod' +import { biometricsDb } from '@/src/db' +import { coordinatorRequest } from '@/src/biometrics/backtest/coordinator-service' +import type { CoordinatorResults } from '@/src/biometrics/backtest/coordinator-tasks' +import { datasetBundleChunks, datasetBundleTarBytes, retainedRawDirectory } from '@/src/biometrics/backtest/dataset-export' +import { ReferenceInputError } from '@/src/biometrics/references/time' + +// One export at a time: each one reads every retained archive of the dataset +// twice (verification, then streaming). A stalled client must not latch it. +let inflight = false +const EXPORT_WATCHDOG_MS = (() => { + const raw = Number(process.env.DATASET_EXPORT_WATCHDOG_MS ?? '') + return Number.isFinite(raw) && raw > 0 ? raw : 30 * 60 * 1000 +})() + +function errorResponse(error: unknown) { + if (error instanceof ReferenceInputError) { + if (error.code === 'NOT_FOUND' || error.code === 'DATASET_RELEASED') return NextResponse.json({ error: 'Dataset not found or released' }, { status: 404 }) + if (error.code === 'DATASET_CORRUPT') return NextResponse.json({ error: `DATASET_CORRUPT: ${error.message}` }, { status: 409 }) + if (error.code === 'WORKER_UNAVAILABLE') return NextResponse.json({ error: error.message }, { status: 503 }) + } + return NextResponse.json({ error: 'Dataset export failed' }, { status: 500 }) +} + +/** + * Stream one frozen calibration dataset as an uncompressed tar: bundle.json + * (index with every file's SHA-256), manifest.json, snapshot.json and the + * retained raw/.RAW.gz archives it references. Nothing else is read. + */ +export async function GET(_request: Request, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params + if (!z.string().uuid().safeParse(id).success) return NextResponse.json({ error: 'Invalid dataset id' }, { status: 400 }) + if (inflight) return NextResponse.json({ error: 'Another dataset export is in progress, retry later' }, { status: 429, headers: { 'Retry-After': '60' } }) + inflight = true + let prepared: CoordinatorResults['exportDataset'] + let directory: string + try { + prepared = await coordinatorRequest(biometricsDb, { type: 'exportDataset', input: { id } }) + directory = retainedRawDirectory(biometricsDb) + } + catch (error) { + inflight = false + return errorResponse(error) + } + + const chunks = datasetBundleChunks(prepared.index, { manifest: prepared.manifestText, snapshot: prepared.snapshotText }, directory) + let watchdog: ReturnType | undefined + const release = () => { + if (watchdog) clearTimeout(watchdog) + inflight = false + } + const body = new ReadableStream({ + start(controller) { + watchdog = setTimeout(() => { + controller.error(new Error('Dataset export exceeded its time limit')) + void chunks.return(undefined).finally(release) + }, EXPORT_WATCHDOG_MS) + }, + async pull(controller) { + try { + const next = await chunks.next() + if (next.done) { + release() + controller.close() + } + else controller.enqueue(next.value) + } + catch (error) { + release() + controller.error(error) + } + }, + async cancel() { + await chunks.return(undefined).finally(release) + }, + }) + return new Response(body, { + headers: { + 'Content-Type': 'application/x-tar', + 'Content-Disposition': `attachment; filename="calibration-dataset-${id}.tar"`, + 'Content-Length': String(datasetBundleTarBytes(prepared.index)), + 'Cache-Control': 'no-store', + 'X-Dataset-Content-Hash': prepared.index.dataset.contentHash, + }, + }) +} diff --git a/docs/biometrics-datasets.md b/docs/biometrics-datasets.md index b52e093a..70fcfb31 100644 --- a/docs/biometrics-datasets.md +++ b/docs/biometrics-datasets.md @@ -249,3 +249,18 @@ reader verification → Python numerical replay → release → collection, whil checking that canonical observations and the parent dataset remain unchanged. Public `signalReplay` remains false until coordinator/API source selection and fenced waveform-job/report integration are wired to this publisher. + +## Off-pod fitting bundle + +`GET /api/biometric-tuning/datasets/{id}/export` streams one live dataset as an +uncompressed tar: `bundle.json` (every other file with its size and SHA-256), +`manifest.json`, `snapshot.json` and the retained `raw/.RAW.gz` archives +the snapshot references. The coordinator verifies the dataset hash, checks that +the live RAW references match the snapshot exactly, and hashes each archive before +anything is sent. The route hashes the archives again while streaming and aborts +on any change. Released or unknown datasets return 404; an integrity failure +returns 409 without streaming. The MCP tool `export_calibration_dataset` returns +the same index and the download path, never waveform bytes. +`scripts/biometrics-local-fit` replays these bundles with the shipped Python +worker and scores them with the shipped evaluator. Its reports are labelled +`exploratory-local` and are never activation evidence. diff --git a/package.json b/package.json index ee79c37d..fdace762 100644 --- a/package.json +++ b/package.json @@ -30,6 +30,7 @@ "prepare": "git rev-parse --git-dir >/dev/null 2>&1 && git config core.hooksPath .githooks || true", "build:biometrics-worker": "node scripts/build/biometrics-worker.mjs", "biometrics:experiment": "tsx scripts/algorithm-experiments/cli.ts", + "biometrics:local-fit": "tsx scripts/biometrics-local-fit/cli.ts", "test:browser": "node tests/browser/run.mjs" }, "lint-staged": { diff --git a/scripts/biometrics-local-fit/README.md b/scripts/biometrics-local-fit/README.md new file mode 100644 index 00000000..c7edc0ae --- /dev/null +++ b/scripts/biometrics-local-fit/README.md @@ -0,0 +1,89 @@ +# Local biometric profile fitting + +Fit biometric profile candidates on a laptop against a frozen calibration +dataset exported from the pod, then send only the best one or two back to the +pod for a real backtest. Waveform replay on Pod 5 runs about 30 times slower +than on a laptop, so parameter sweeps belong here. Activation decisions do not: +everything this tool writes is labelled `exploratory-local` and +`not_activation_evidence`. + +## 1. Export the dataset + +```sh +curl -fo dataset.tar http://POD:3000/api/biometric-tuning/datasets/DATASET_ID/export +``` + +The MCP tool `export_calibration_dataset` returns the same bundle index and this +download path. Waveform bytes never go through MCP. The tar holds: + +- `bundle.json`: the index. It lists every other file with its exact size and SHA-256. +- `manifest.json` and `snapshot.json`: the frozen dataset. Together they hash + to the dataset's `contentHash`. +- `raw/.RAW.gz`: each retained archive the snapshot references, and nothing else. + +A released dataset returns 404. If a retained archive fails its hash check on +the pod, the export returns 409 and streams nothing. Archives contain both +sides of the bed; keep the bundle on a machine you trust. + +## 2. Set up the replay environment + +```sh +(cd modules/piezo-processor && uv sync --frozen) +``` + +This installs the locked Python 3.10, NumPy and SciPy that the pod uses. The +runner calls the unchanged `common.biometrics.replay_worker.run()` through +`local_replay.py`. The only differences from the pod are that it skips the +pod's host-wide execution lock, CPU rlimit and nice level, so nights can replay +in parallel. `run.json` records the local Python build identity. Its +`identity.sources` hashes can be compared with the pod's +(`python -B -m common.biometrics.replay_worker --describe-build`). The +interpreter and wheel hashes differ by platform by design. + +## 3. Fit or sweep + +```sh +pnpm biometrics:local-fit verify dataset.tar +pnpm biometrics:local-fit fit dataset.tar candidates.json out/ +pnpm biometrics:local-fit sweep dataset.tar default out/ # or a GRID.json +``` + +`candidates.json` is `[{ "id": "tol12", "overrides": [{ "path": "heart_rate.trackingToleranceBpm", "value": 12 }] }]`. +The shipped defaults are always added as `defaults`. A grid maps override paths +to value lists. A comma-joined key varies several paths together, for example +`"breathing_rate.intervalLowerRatio,breathing_rate.intervalUpperRatio": [[0.5, 1.5], [0.4, 1.7]]`. +The built-in `default` grid has 18 points, sized to finish in under 15 minutes +on a 10-core laptop for seven frozen nights. + +The tool verifies every file hash, the dataset content hash and the exact set +of archives before any replay. It then replays each tuning and validation night +once per distinct configuration, longest nights first, across `--jobs` workers +(default: all cores). Each candidate and split is scored with the pod's own +`evaluateBacktest`, against defaults on common support. Holdout nights are +never replayed locally: holdout runs once, on the pod, for the final candidate. + +Outputs: + +- `out//.json`: per-night HR, breathing and HRV error, plus + the full evaluator report. A metric without paired reference data reads + `no_paired_reference`, never zero error. +- `out/ranking.json` and `out/ranking.md`: ranked by the tuning split's + equal-night mean HR MAE difference from defaults, with validation alongside. +- `out/run.json`: timing, worker count and the Python build identity. + +## 4. Check parity, then confirm on the pod + +```sh +curl -fo pod-validation.json http://POD:3000/api/biometric-tuning/backtests/JOB_ID/result +pnpm biometrics:local-fit parity out/ pod-validation.json [pod-tuning.json] +``` + +`parity` compares local per-night HR error for defaults with the pod's +results. It exits non-zero if any night differs by more than 0.1 bpm. + +For the top one or two candidates, create a revision with +`propose_biometric_profile`, using the same overrides. Its `resolvedHash` +matches the local report. Then run `backtest_biometrics` on tuning and +validation. Run holdout once, for the final candidate only. Only the pod's +report can back activation, and applying a profile still needs the user's +explicit authorization. diff --git a/scripts/biometrics-local-fit/bundle.test.ts b/scripts/biometrics-local-fit/bundle.test.ts new file mode 100644 index 00000000..207f335d --- /dev/null +++ b/scripts/biometrics-local-fit/bundle.test.ts @@ -0,0 +1,105 @@ +// @vitest-environment node +import { createWriteStream, mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from 'node:fs' +import { once } from 'node:events' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, expect, it } from 'vitest' +import { TAR_END, tarHeader, tarPadding } from '@/src/biometrics/backtest/bundle-tar' +import { datasetBundleChunks, prepareDatasetExport } from '@/src/biometrics/backtest/dataset-export' +import { exportableWaveformDataset } from '@/src/biometrics/backtest/tests/fixtures/export' +import { extractBundle, verifyBundle } from './bundle' + +let directory: string +let fixture: ReturnType +let tarPath: string +beforeEach(async () => { + directory = mkdtempSync(join(tmpdir(), 'local-fit-bundle-')) + fixture = exportableWaveformDataset(directory) + const prepared = await prepareDatasetExport(fixture.db, fixture.id) + tarPath = join(directory, 'bundle.tar') + const out = createWriteStream(tarPath) + for await (const chunk of datasetBundleChunks(prepared.index, { manifest: prepared.manifestText, snapshot: prepared.snapshotText }, fixture.rawDirectory)) out.write(chunk) + out.end() + await once(out, 'finish') +}) +afterEach(() => { + fixture.sqlite.close() + rmSync(directory, { recursive: true, force: true }) +}) + +const extract = () => extractBundle(tarPath, join(directory, `out-${Math.random().toString(16).slice(2)}`)) +const flip = (path: string) => { + const bytes = readFileSync(path) + bytes[bytes.length - 1] ^= 0xff + writeFileSync(path, bytes) +} + +it('extracts and verifies an exported bundle end to end', async () => { + const bundle = await verifyBundle(await extract()) + expect(bundle.index.dataset.contentHash).toBe(fixture.contentHash) + expect(bundle.snapshot.version).toBe('waveform-dataset-v1') + expect(bundle.manifest.splitProtocol.assignments.map(row => row.split)).toEqual(['tuning', 'validation', 'holdout']) +}) + +it('rejects a RAW archive whose bytes do not match the index', async () => { + const root = await extract() + flip(join(root, 'raw', fixture.artifacts[1].name)) + await expect(verifyBundle(root)).rejects.toThrow(`SHA-256 mismatch: raw/${fixture.artifacts[1].name}`) +}) + +it('rejects edited snapshots even when the index is rewritten to match', async () => { + const root = await extract() + const snapshot = JSON.parse(readFileSync(join(root, 'snapshot.json'), 'utf8')) + snapshot.subjectId = 'someone else' + const text = JSON.stringify(snapshot) + writeFileSync(join(root, 'snapshot.json'), text) + const index = JSON.parse(readFileSync(join(root, 'bundle.json'), 'utf8')) + const { createHash } = await import('node:crypto') + Object.assign(index.files[1], { sha256: createHash('sha256').update(text).digest('hex'), bytes: Buffer.byteLength(text) }) + writeFileSync(join(root, 'bundle.json'), JSON.stringify(index)) + await expect(verifyBundle(root)).rejects.toThrow('content hash') +}) + +it('rejects files the index does not list, and missing or linked files', async () => { + const root = await extract() + writeFileSync(join(root, 'raw', fixture.unrelated.name), 'extra') + await expect(verifyBundle(root)).rejects.toThrow('unlisted') + rmSync(join(root, 'raw', fixture.unrelated.name)) + const raw = join(root, 'raw', fixture.artifacts[0].name) + rmSync(raw) + await expect(verifyBundle(root)).rejects.toThrow('missing') + symlinkSync(join(fixture.rawDirectory, fixture.artifacts[0].name), raw) + await expect(verifyBundle(root)).rejects.toThrow('not a regular file') +}) + +it('rejects a RAW set that differs from what the frozen snapshot references', async () => { + const root = await extract() + const index = JSON.parse(readFileSync(join(root, 'bundle.json'), 'utf8')) + index.files = index.files.filter((file: { path: string }) => file.path !== `raw/${fixture.artifacts[2].name}`) + writeFileSync(join(root, 'bundle.json'), JSON.stringify(index)) + rmSync(join(root, 'raw', fixture.artifacts[2].name)) + await expect(verifyBundle(root)).rejects.toThrow('not exactly the frozen dataset') +}) + +it('refuses tar entries that are unlisted, resized, links or outside the bundle', async () => { + const index = readFileSync(tarPath).subarray(512, 512 + Number.parseInt(readFileSync(tarPath).subarray(124, 135).toString(), 8)) + const write = (entries: Buffer[]) => { + tarPath = join(directory, `crafted-${Math.random().toString(16).slice(2)}.tar`) + writeFileSync(tarPath, Buffer.concat([tarHeader('bundle.json', index.length), index, tarPadding(index.length), ...entries, TAR_END])) + } + write([tarHeader('notes.txt', 2), Buffer.from('hi'), tarPadding(2)]) + await expect(extract()).rejects.toThrow('Unlisted or resized bundle entry: notes.txt') + write([tarHeader('manifest.json', 1), Buffer.from('x'), tarPadding(1)]) + await expect(extract()).rejects.toThrow('Unlisted or resized') + const link = tarHeader('manifest.json', 0) + link[156] = '2'.charCodeAt(0) + let sum = 0 + link.fill(32, 148, 156) + for (const byte of link) sum += byte + link.write(`${sum.toString(8).padStart(6, '0')}\0 `, 148, 'ascii') + write([link]) + await expect(extract()).rejects.toThrow('only regular files') + expect(() => tarHeader('../escape', 1)).toThrow('Invalid bundle entry name') + write([]) + await expect(extract()).rejects.toThrow('missing') +}) diff --git a/scripts/biometrics-local-fit/bundle.ts b/scripts/biometrics-local-fit/bundle.ts new file mode 100644 index 00000000..73954123 --- /dev/null +++ b/scripts/biometrics-local-fit/bundle.ts @@ -0,0 +1,110 @@ +import { createHash } from 'node:crypto' +import { constants, createReadStream, createWriteStream } from 'node:fs' +import { lstat, mkdir, open, readdir, readFile, stat, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { pipeline } from 'node:stream/promises' +import { dataHash } from '@/src/biometrics/algorithms/canonical' +import { parseTarHeader } from '@/src/biometrics/backtest/bundle-tar' +import { datasetManifestSchema, datasetSnapshotSchema } from '@/src/biometrics/backtest/contracts' +import { DATASET_BUNDLE_INDEX_PATH, datasetBundleIndexSchema, type DatasetBundleIndex } from '@/src/biometrics/backtest/dataset-export' + +const MAX_INDEX_BYTES = 1024 * 1024 +const MAX_TEXT_BYTES = 16 * 1024 * 1024 +const MAX_RAW_BYTES = 256 * 1024 * 1024 + +export class BundleError extends Error {} + +async function sha256File(path: string) { + const digest = createHash('sha256') + for await (const chunk of createReadStream(path, { highWaterMark: 1024 * 1024 })) digest.update(chunk as Buffer) + return digest.digest('hex') +} + +async function readExactly(handle: Awaited>, position: number, length: number) { + const buffer = Buffer.alloc(length) + const { bytesRead } = await handle.read(buffer, 0, length, position) + if (bytesRead !== length) throw new BundleError('Truncated bundle archive') + return buffer +} + +/** Unpack an export tar into a new directory. Only the index and the files it + * lists are accepted, each once and at its listed size; verifyBundle then + * checks every hash. */ +export async function extractBundle(tarPath: string, destination: string) { + await mkdir(destination, { recursive: false, mode: 0o700 }) + await mkdir(join(destination, 'raw'), { mode: 0o700 }) + const handle = await open(tarPath, constants.O_RDONLY) + try { + const total = (await handle.stat()).size + let position = 0, index: DatasetBundleIndex | undefined + const expected = new Map(), seen = new Set() + for (;;) { + const header = parseTarHeader(await readExactly(handle, position, 512)) + position += 512 + if (!header) break + if (seen.has(header.name)) throw new BundleError(`Duplicate bundle entry: ${header.name}`) + seen.add(header.name) + if (!index) { + if (header.name !== DATASET_BUNDLE_INDEX_PATH || header.bytes > MAX_INDEX_BYTES) throw new BundleError('Bundle must start with its bundle.json index') + const text = await readExactly(handle, position, header.bytes) + index = datasetBundleIndexSchema.parse(JSON.parse(text.toString('utf8'))) + for (const file of index.files) expected.set(file.path, file.bytes) + await writeFile(join(destination, DATASET_BUNDLE_INDEX_PATH), text, { flag: 'wx', mode: 0o600 }) + } + else { + if (expected.get(header.name) !== header.bytes) throw new BundleError(`Unlisted or resized bundle entry: ${header.name}`) + if (position + header.bytes > total) throw new BundleError('Truncated bundle archive') + await pipeline(createReadStream('', { fd: handle.fd, autoClose: false, start: position, end: position + header.bytes - 1 }), + createWriteStream(join(destination, header.name), { flags: 'wx', mode: 0o600 })) + } + position += Math.ceil(header.bytes / 512) * 512 + } + if (!index) throw new BundleError('Bundle has no index') + const missing = index.files.filter(file => !seen.has(file.path)) + if (missing.length) throw new BundleError(`Bundle is missing ${missing.length} listed file(s), e.g. ${missing[0].path}`) + } + finally { + await handle.close() + } + return destination +} + +/** Check a bundle directory against its index: exact file set, sizes and + * SHA-256s, the dataset's own content hash, and that the RAW archives are + * exactly the ones the frozen snapshot references. */ +export async function verifyBundle(directory: string) { + const indexPath = join(directory, DATASET_BUNDLE_INDEX_PATH) + if ((await stat(indexPath)).size > MAX_INDEX_BYTES) throw new BundleError('Bundle index is too large') + const index = datasetBundleIndexSchema.parse(JSON.parse(await readFile(indexPath, 'utf8'))) + const listed = new Set(index.files.map(file => file.path)) + if (listed.size !== index.files.length) throw new BundleError('Bundle index lists a file twice') + const present = [ + ...(await readdir(directory)).filter(name => name !== 'raw'), + ...(await readdir(join(directory, 'raw')).catch(() => [] as string[])).map(name => `raw/${name}`), + ] + const extra = present.filter(name => name !== DATASET_BUNDLE_INDEX_PATH && !listed.has(name)) + if (extra.length) throw new BundleError(`Bundle contains unlisted file(s), e.g. ${extra[0]}`) + for (const file of index.files) { + const path = join(directory, file.path) + const info = await lstat(path).catch(() => null) + if (!info?.isFile()) throw new BundleError(`Bundle file is missing or not a regular file: ${file.path}`) + const limit = file.role === 'raw' ? MAX_RAW_BYTES : MAX_TEXT_BYTES + if (info.size !== file.bytes || info.size > limit) throw new BundleError(`Bundle file size mismatch: ${file.path}`) + if (await sha256File(path) !== file.sha256) throw new BundleError(`Bundle file SHA-256 mismatch: ${file.path}`) + } + const text = (role: 'manifest' | 'snapshot') => readFile(join(directory, index.files.find(file => file.role === role)?.path ?? ''), 'utf8') + const manifest = datasetManifestSchema.parse(JSON.parse(await text('manifest'))) + const snapshot = datasetSnapshotSchema.parse(JSON.parse(await text('snapshot'))) + if (dataHash({ manifest, snapshot }) !== index.dataset.contentHash) throw new BundleError('Dataset content hash does not match its manifest and snapshot') + if (snapshot.side !== index.dataset.side || snapshot.version !== index.dataset.snapshotVersion) throw new BundleError('Bundle index does not describe this snapshot') + const frozen = new Map(snapshot.version === 'waveform-dataset-v1' + ? snapshot.waveformInputs.flatMap(cohort => cohort.input.artifacts.map(source => [source.artifact.sha256, source.artifact] as const)) + : []) + const raw = index.files.filter(file => file.role === 'raw') + if (raw.length !== frozen.size || raw.some((file) => { + const artifact = frozen.get(file.sha256) + return !artifact || artifact.compressed_bytes !== file.bytes || artifact.expanded_bytes !== file.expandedBytes + })) throw new BundleError('Bundle RAW archives are not exactly the frozen dataset\'s archives') + return { index, manifest, snapshot, rawDirectory: join(directory, 'raw') } +} +export type VerifiedBundle = Awaited> diff --git a/scripts/biometrics-local-fit/cli.ts b/scripts/biometrics-local-fit/cli.ts new file mode 100644 index 00000000..08017583 --- /dev/null +++ b/scripts/biometrics-local-fit/cli.ts @@ -0,0 +1,106 @@ +import { mkdir, readFile, rm, stat, writeFile } from 'node:fs/promises' +import { join } from 'node:path' +import { pathToFileURL } from 'node:url' +import { extractBundle, verifyBundle } from './bundle' +import { candidateFileSchema, defaultPythonLocation, fitCandidates, LOCAL_SPLITS, type Candidate, type LocalSplit } from './fit' +import { DEFAULT_GRID, expandGrid } from './grid' +import { compareWithPod, parseLocalReport, rankCandidates, rankingMarkdown } from './report' + +const usage = `Usage: + verify BUNDLE + fit BUNDLE CANDIDATES.json OUTPUT_DIR [--jobs N] [--splits tuning,validation] [--protocol FILE] [--python PATH] + sweep BUNDLE GRID.json|default OUTPUT_DIR [same options as fit] + parity OUTPUT_DIR POD_RESULT.json [POD_RESULT.json ...] +BUNDLE is the tar from GET /api/biometric-tuning/datasets/{id}/export, or a directory extracted from it.` + +function options(args: string[]) { + const values: Record = {} + const positional: string[] = [] + for (let index = 0; index < args.length; index++) { + if (args[index].startsWith('--')) values[args[index].slice(2)] = args[++index] ?? '' + else positional.push(args[index]) + } + return { values, positional } +} + +async function readJson(path: string) { + return JSON.parse(await readFile(path, 'utf8')) as unknown +} + +async function openBundle(path: string, outputDirectory: string) { + if ((await stat(path)).isDirectory()) return verifyBundle(path) + const destination = join(outputDirectory, 'bundle') + await rm(destination, { recursive: true, force: true }) + await extractBundle(path, destination) + return verifyBundle(destination) +} + +async function fit(bundlePath: string, candidates: Candidate[], outputDirectory: string, flags: Record) { + await mkdir(outputDirectory, { recursive: true }) + const bundle = await openBundle(bundlePath, outputDirectory) + const splits = (flags.splits?.split(',') ?? [...LOCAL_SPLITS]) as LocalSplit[] + const started = performance.now() + const result = await fitCandidates({ bundle, candidates, outputDirectory, splits, + jobs: flags.jobs ? Number(flags.jobs) : undefined, + protocol: flags.protocol ? await readJson(flags.protocol) as never : undefined, + python: flags.python ? defaultPythonLocation(flags.python) : undefined, + onProgress: (done, total, item) => process.stderr.write(`[${done}/${total}] ${item.cohortId} ${item.candidateIds.join(',') || '(baseline)'} ${(item.wallMs / 1000).toFixed(1)}s\n`) }) + for (const report of result.reports) { + await mkdir(join(outputDirectory, report.candidate.id), { recursive: true }) + await writeFile(join(outputDirectory, report.candidate.id, `${report.split}.json`), `${JSON.stringify(report)}\n`) + } + const ranking = rankCandidates(result.reports) + await writeFile(join(outputDirectory, 'ranking.json'), `${JSON.stringify(ranking, null, 2)}\n`) + await writeFile(join(outputDirectory, 'ranking.md'), rankingMarkdown(ranking)) + const totalWallMs = performance.now() - started + await writeFile(join(outputDirectory, 'run.json'), `${JSON.stringify({ version: 'biometrics-local-fit-run-v1', label: 'exploratory-local', + dataset: bundle.index.dataset, splits, candidates: ranking.rows.length, workItems: result.workItems, jobs: result.jobs, + replayWallMs: result.replayWallMs, totalWallMs, pythonBuild: { hash: result.build.hash, sourcesHash: result.build.sourcesHash, identity: result.build.identity } }, null, 2)}\n`) + await rm(join(outputDirectory, 'work'), { recursive: true, force: true }) + process.stdout.write(rankingMarkdown(ranking)) + process.stdout.write(`\n${ranking.rows.length} candidates, ${result.workItems} replays on ${result.jobs} workers in ${(totalWallMs / 60000).toFixed(1)} min\n`) +} + +export async function runLocalFitCli(args: string[]) { + const [command, ...rest] = args + const { values, positional } = options(rest) + if (command === 'verify' && positional.length === 1) { + const path = positional[0] + let bundle + if ((await stat(path)).isDirectory()) bundle = await verifyBundle(path) + else { + const directory = `${path}.verify-${process.pid}` + try { + bundle = await verifyBundle(await extractBundle(path, directory)) + } + finally { + await rm(directory, { recursive: true, force: true }) + } + } + process.stdout.write(`${JSON.stringify({ verified: true, dataset: bundle.index.dataset, files: bundle.index.files.length })}\n`) + return + } + if (command === 'fit' && positional.length === 3) { + return fit(positional[0], candidateFileSchema.parse(await readJson(positional[1])), positional[2], values) + } + if (command === 'sweep' && positional.length === 3) { + const grid = positional[1] === 'default' ? DEFAULT_GRID : await readJson(positional[1]) + return fit(positional[0], expandGrid(grid), positional[2], values) + } + if (command === 'parity' && positional.length >= 2) { + const defaults = await Promise.all(LOCAL_SPLITS.map(split => readJson(join(positional[0], 'defaults', `${split}.json`)).then(parseLocalReport).catch(() => null))) + const parity = compareWithPod(defaults.filter(row => row !== null), await Promise.all(positional.slice(1).map(readJson))) + await writeFile(join(positional[0], 'parity.json'), `${JSON.stringify(parity, null, 2)}\n`) + process.stdout.write(`${JSON.stringify(parity, null, 2)}\n`) + if (!parity.allWithinTolerance) process.exitCode = 2 + return + } + throw new Error(usage) +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + runLocalFitCli(process.argv.slice(2)).catch((error: unknown) => { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`) + process.exitCode = 1 + }) +} diff --git a/scripts/biometrics-local-fit/fit.ts b/scripts/biometrics-local-fit/fit.ts new file mode 100644 index 00000000..2238db89 --- /dev/null +++ b/scripts/biometrics-local-fit/fit.ts @@ -0,0 +1,172 @@ +import { execFile } from 'node:child_process' +import { copyFile, link, mkdir, readdir, writeFile } from 'node:fs/promises' +import { availableParallelism, tmpdir } from 'node:os' +import { join, resolve } from 'node:path' +import { promisify } from 'node:util' +import { z } from 'zod' +import { configHash, dataHash } from '@/src/biometrics/algorithms/canonical' +import { defaultConfig, manifestHash, resolveConfig, type BiometricConfig, type Override } from '@/src/biometrics/algorithms/registry' +import { experimentProtocolSchema } from '@/src/biometrics/backtest/contracts' +import { evaluateBacktest } from '@/src/biometrics/backtest/evaluator' +import { waveformPythonBuildSchema } from '@/src/biometrics/backtest/waveform-build' +import { runWaveformChild } from '@/src/biometrics/backtest/waveform-child' +import type { waveformEvaluationSchema } from '@/src/biometrics/backtest/waveform-output' +import type { VerifiedBundle } from './bundle' +import { summarizeLocalReport, type LocalFitReport } from './report' + +export const LOCAL_SPLITS = ['tuning', 'validation'] as const +export type LocalSplit = typeof LOCAL_SPLITS[number] +/** Matches the protocol of the pod experiment the frozen nights were scored with. */ +export const DEFAULT_PROTOCOL = { version: 'descriptive-evaluation-v1', mode: 'signal_replay', assumedStageCalibrationQuality: 0, pointToleranceMs: 30000, clockOffsetMs: 0 } as const +const LOCAL_EXPERIMENT_ID = '00000000-0000-4000-8000-000000000000' +const REPO = resolve(import.meta.dirname, '../..') + +export const candidateSchema = z.object({ id: z.string().regex(/^[A-Za-z0-9_.-]{1,80}$/), + overrides: z.array(z.object({ path: z.string().min(1).max(200), value: z.unknown() }).strict()).max(64) }).strict() +export const candidateFileSchema = z.array(candidateSchema).min(1).max(500) + .refine(rows => new Set(rows.map(row => row.id)).size === rows.length, 'Candidate ids must be unique') +export type Candidate = z.infer + +export interface PythonLocation { pythonPath: string, modulesPath: string, script: string } +export function defaultPythonLocation(pythonPath?: string): PythonLocation { + return { pythonPath: pythonPath ?? join(REPO, 'modules/piezo-processor/.venv/bin/python'), modulesPath: join(REPO, 'modules'), + script: join(REPO, 'scripts/biometrics-local-fit/local_replay.py') } +} + +export async function describeLocalBuild(location: PythonLocation) { + const { stdout } = await promisify(execFile)(location.pythonPath, ['-B', location.script, '--describe-build', '-'], { + cwd: tmpdir(), env: { NODE_ENV: 'production', PYTHONPATH: location.modulesPath, LANG: 'C.UTF-8', TZ: 'UTC' }, encoding: 'utf8', maxBuffer: 65536, timeout: 30000, + }).catch(() => { + throw new Error(`Cannot run ${location.pythonPath}; create it with: (cd modules/piezo-processor && uv sync --frozen)`) + }) + const build = waveformPythonBuildSchema.parse(JSON.parse(stdout)) + if (configHash(build.identity) !== build.hash) throw new Error('Local Python build identity integrity mismatch') + return { ...build, sourcesHash: configHash(build.identity.sources) } +} + +/** The read-only artifact store serializes readers on its directory lock, so + * each worker slot gets a private directory of hard links (copies across devices). */ +export async function prepareWorkerSlots(rawDirectory: string, workDirectory: string, count: number) { + const names = (await readdir(rawDirectory)).filter(name => /^[a-f0-9]{64}\.RAW\.gz$/.test(name)) + const slots: string[] = [] + for (let index = 0; index < count; index++) { + const slot = join(workDirectory, `slot-${index}`) + await mkdir(slot, { recursive: true, mode: 0o700 }) + for (const name of names) { + await link(join(rawDirectory, name), join(slot, name)).catch(async (error: NodeJS.ErrnoException) => { + if (error.code === 'EEXIST') return + if (error.code !== 'EXDEV') throw error + await copyFile(join(rawDirectory, name), join(slot, name)) + }) + } + await writeFile(join(slot, '.lock'), '', { mode: 0o600 }) + slots.push(slot) + } + return slots +} + +async function runPool(items: T[], slots: string[], work: (item: T, slot: string) => Promise) { + let next = 0, failed = false + await Promise.all(slots.map(async (slot) => { + while (!failed && next < items.length) { + try { + await work(items[next++], slot) + } + catch (error) { + failed = true + throw error + } + } + })) +} + +export interface ResolvedCandidate extends Candidate { resolved: BiometricConfig, resolvedHash: string } +export function resolveCandidates(candidates: Candidate[]): ResolvedCandidate[] { + const rows = candidates.some(candidate => candidate.id === 'defaults') ? candidates : [{ id: 'defaults', overrides: [] }, ...candidates] + return rows.map((candidate) => { + const { resolved, hash } = resolveConfig(candidate.overrides as Override[]) + return { ...candidate, resolved: resolved as BiometricConfig, resolvedHash: hash } + }) +} + +export interface FitOptions { + bundle: VerifiedBundle + candidates: Candidate[] + outputDirectory: string + splits?: readonly LocalSplit[] + jobs?: number + protocol?: z.input + python?: PythonLocation + onProgress?: (done: number, total: number, item: { cohortId: string, candidateIds: string[], wallMs: number }) => void +} + +/** Replay each selected night once per distinct resolved configuration, then + * score each candidate and split with the pod's own evaluator. Holdout nights + * are never replayed here. */ +export async function fitCandidates(options: FitOptions) { + const { bundle } = options + if (bundle.snapshot.version !== 'waveform-dataset-v1') throw new Error('Local fitting needs a retained waveform dataset') + const snapshot = bundle.snapshot + const splits = options.splits ?? LOCAL_SPLITS + if (splits.some(split => !LOCAL_SPLITS.includes(split))) throw new Error('Holdout runs once, on the pod, for the final candidate only') + const protocol = experimentProtocolSchema.parse(options.protocol ?? DEFAULT_PROTOCOL) + if (protocol.mode !== 'signal_replay') throw new Error('Local fitting scores waveform replay; use a signal_replay protocol') + const python = options.python ?? defaultPythonLocation() + const build = await describeLocalBuild(python) + const candidates = resolveCandidates(options.candidates) + const defaults = { manifestHash, resolvedHash: configHash(defaultConfig), resolved: defaultConfig as BiometricConfig } + const configs = new Map([[defaults.resolvedHash, defaults.resolved], ...candidates.map(row => [row.resolvedHash, row.resolved] as const)]) + const assignments = bundle.manifest.splitProtocol.assignments.filter(row => (splits as readonly string[]).includes(row.split)) + const inputs = new Map(snapshot.waveformInputs.map(row => [row.cohortId, row.input])) + const items = assignments.flatMap(assignment => [...configs.keys()].map(hash => ({ cohortId: assignment.cohortId, hash }))) + const length = (cohortId: string) => { + const input = inputs.get(cohortId) + if (!input) throw new Error(`Bundle has no retained input for ${cohortId}`) + return input.score_end_ms - input.retained_start_ms + } + // Longest nights first keeps every slot busy until the end. + items.sort((a, b) => length(b.cohortId) - length(a.cohortId)) + const jobs = Math.max(1, Math.min(options.jobs ?? availableParallelism(), items.length)) + const work = join(options.outputDirectory, 'work') + const slots = await prepareWorkerSlots(bundle.rawDirectory, work, jobs) + const outputs = new Map['outputs'][number]>() + const otherSideSide = snapshot.side === 'left' ? 'right' : 'left' + const started = performance.now() + let done = 0 + await runPool(items, slots, async (item, slot) => { + const input = inputs.get(item.cohortId) + const resolved = configs.get(item.hash) + if (!input || !resolved) throw new Error('Missing replay input') + const itemStarted = performance.now() + const output = await runWaveformChild({ ...python, retainedDirectory: slot }, + { version: 'retained-waveform-work-v1', input, configs: { [snapshot.side]: resolved, [otherSideSide]: defaults.resolved } }, + { timeoutMs: 3600000, expectedBuildHash: build.hash }) + outputs.set(`${item.cohortId}:${item.hash}`, { cohortId: item.cohortId, configHash: item.hash, ...output }) + options.onProgress?.(++done, items.length, { cohortId: item.cohortId, + candidateIds: candidates.filter(row => row.resolvedHash === item.hash).map(row => row.id), wallMs: performance.now() - itemStarted }) + }) + const replayWallMs = performance.now() - started + const evaluatorBuildHash = configHash({ version: 'biometrics-local-fit-evaluator-v1', pythonBuildHash: build.hash, node: process.version }) + const reports: LocalFitReport[] = [] + for (const candidate of candidates) { + for (const split of splits) { + const cohorts = assignments.filter(row => row.split === split).map(row => row.cohortId) + if (!cohorts.length) continue + const jobConfigs = [{ label: 'default' as const, revisionId: null, ...defaults }, + { label: 'candidate' as const, revisionId: null, manifestHash, resolvedHash: candidate.resolvedHash, resolved: candidate.resolved }] + const runOutputs = cohorts.flatMap(cohortId => [...new Set(jobConfigs.map(config => config.resolvedHash))].map((hash) => { + const output = outputs.get(`${cohortId}:${hash}`) + if (!output) throw new Error(`Missing replay output for ${cohortId}`) + return output + })) + const job = { version: 'backtest-job-v1' as const, experimentId: LOCAL_EXPERIMENT_ID, idempotencyKey: `local:${candidate.id}:${split}`, + evaluatorBuildHash, split, configs: jobConfigs, waveform: { pythonBuildHash: build.hash, otherSide: defaults }, + datasetId: bundle.index.dataset.id, datasetHash: bundle.index.dataset.contentHash, protocol, protocolHash: configHash(protocol), + holdoutUse: 0, evidence: 'descriptive_only' as const } + const report = evaluateBacktest({ job, dataset: { manifest: bundle.manifest, snapshot }, waveform: { hash: dataHash(runOutputs), outputs: runOutputs } }) + reports.push(summarizeLocalReport({ candidate, split, report, dataset: bundle.index.dataset, + build: { pythonBuildHash: build.hash, pythonSourcesHash: build.sourcesHash, evaluatorBuildHash }, protocol })) + } + } + return { reports, build, replayWallMs, workItems: items.length, jobs } +} diff --git a/scripts/biometrics-local-fit/grid.ts b/scripts/biometrics-local-fit/grid.ts new file mode 100644 index 00000000..c95132c6 --- /dev/null +++ b/scripts/biometrics-local-fit/grid.ts @@ -0,0 +1,40 @@ +import { z } from 'zod' +import type { Candidate } from './fit' + +/** Axis keys are override paths; a comma-joined key varies several paths + * together (for example both breathing ratios), each value then an array. */ +export const gridSchema = z.record(z.string().min(1).max(400), z.array(z.unknown()).min(1).max(20)) + .refine(grid => Object.keys(grid).length > 0 && Object.keys(grid).length <= 8, 'Use 1-8 grid axes') + +/** Sized to finish under 15 minutes on a 10-core laptop for the seven frozen + * nights (tuning + validation is about 28 night-hours per candidate). HR error + * depends only on the two heart-rate parameters; the breathing ratios and + * minimumWindows axis is varied jointly so the grid still covers them. */ +export const DEFAULT_GRID = { + 'heart_rate.trackingToleranceBpm': [10, 15, 20], + 'heart_rate.trackingHistoryLength': [3, 5, 8], + 'breathing_rate.intervalLowerRatio,breathing_rate.intervalUpperRatio,hrv_index.minimumWindows': [[0.5, 1.5, 10], [0.4, 1.7, 20]], +} satisfies z.input + +const label = (value: unknown) => String(value).replace(/[^A-Za-z0-9.]/g, '') +const short = (path: string) => ({ + 'heart_rate.trackingToleranceBpm': 'tol', 'heart_rate.trackingHistoryLength': 'hist', + 'breathing_rate.intervalLowerRatio': 'brlo', 'breathing_rate.intervalUpperRatio': 'brhi', + 'hrv_index.harmonicTolerance': 'harm', 'hrv_index.minimumWindows': 'mw', +} as Record)[path] ?? path.split('.').at(-1) ?? path + +export function expandGrid(raw: unknown): Candidate[] { + const grid = gridSchema.parse(raw) + const axes = Object.entries(grid).map(([key, values]) => { + const paths = key.split(',') + return values.map((value) => { + const parts = paths.length === 1 ? [value] : value + if (!Array.isArray(parts) || parts.length !== paths.length) throw new Error(`Grid axis ${key} needs ${paths.length} values per entry`) + return paths.map((path, index) => ({ path, value: parts[index] })) + }) + }) + let combos: Array> = [[]] + for (const axis of axes) combos = combos.flatMap(combo => axis.map(level => [...combo, ...level])) + if (combos.length > 500) throw new Error('Grid expands to more than 500 candidates') + return combos.map(overrides => ({ id: overrides.map(row => `${short(row.path)}${label(row.value)}`).join('-').slice(0, 80), overrides })) +} diff --git a/scripts/biometrics-local-fit/local_replay.py b/scripts/biometrics-local-fit/local_replay.py new file mode 100644 index 00000000..433084b7 --- /dev/null +++ b/scripts/biometrics-local-fit/local_replay.py @@ -0,0 +1,34 @@ +"""Laptop entry point for the shipped retained waveform worker. + +Calls the unchanged common.biometrics.replay_worker.run(), so readings come +from the same sources the pod pins in its Python build hash. Only the pod's +host-wide execution lock, CPU rlimit and nice level are left out, which lets +several nights replay in parallel on a development machine. +""" +import os +import sys + +for key in ('OPENBLAS_NUM_THREADS', 'OMP_NUM_THREADS', 'MKL_NUM_THREADS', 'NUMEXPR_NUM_THREADS'): + os.environ[key] = '1' + + +def main(): + if len(sys.argv) != 3 or sys.argv[1] not in ('--retained-directory', '--describe-build'): + sys.stderr.write('usage: local_replay.py --retained-directory DIR | --describe-build -\n') + return 2 + try: + if sys.argv[1] == '--describe-build': + import json + from common.biometrics.build_identity import replay_build_identity + sys.stdout.write(json.dumps(replay_build_identity(), allow_nan=False, separators=(',', ':')) + '\n') + else: + from common.biometrics.replay_worker import run + run(sys.stdin.buffer, sys.stdout.buffer, sys.argv[2]) + return 0 + except Exception: + sys.stderr.write('Retained waveform worker failed\n') + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/scripts/biometrics-local-fit/report.test.ts b/scripts/biometrics-local-fit/report.test.ts new file mode 100644 index 00000000..136fd18c --- /dev/null +++ b/scripts/biometrics-local-fit/report.test.ts @@ -0,0 +1,118 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' +import { configHash, dataHash } from '@/src/biometrics/algorithms/canonical' +import { defaultConfig, manifestHash, resolveConfig } from '@/src/biometrics/algorithms/registry' +import { evaluateBacktest } from '@/src/biometrics/backtest/evaluator' +import { exportableWaveformDataset, exportNights } from '@/src/biometrics/backtest/tests/fixtures/export' +import { DEFAULT_PROTOCOL, fitCandidates } from './fit' +import { DEFAULT_GRID, expandGrid } from './grid' +import { compareWithPod, parseLocalReport, rankCandidates, rankingMarkdown, summarizeLocalReport } from './report' + +const pythonBuildHash = 'b'.repeat(64) +let directory: string +let fixture: ReturnType +beforeAll(() => { + directory = mkdtempSync(join(tmpdir(), 'local-fit-report-')) + fixture = exportableWaveformDataset(directory) +}) +afterAll(() => { + fixture.sqlite.close() + rmSync(directory, { recursive: true, force: true }) +}) + +/** Score a candidate with fixed synthetic readings (references are 62 bpm). */ +function localReport(id: string, overrides: Array<{ path: string, value: unknown }>, split: 'tuning' | 'validation', heartRates: { default: number, candidate: number }) { + const defaults = { manifestHash, resolvedHash: configHash(defaultConfig), resolved: defaultConfig } + const candidate = resolveConfig(overrides as never) + const configs = [{ label: 'default' as const, revisionId: null, ...defaults }, + { label: 'candidate' as const, revisionId: null, manifestHash, resolvedHash: candidate.hash, resolved: candidate.resolved }] + const night = exportNights.find(row => row.split === split) ?? exportNights[0] + const artifact = fixture.snapshot.version === 'waveform-dataset-v1' ? fixture.snapshot.waveformInputs.find(row => row.cohortId === night.cohortId)?.input.artifacts[0].artifact.sha256 ?? '' : '' + const outputs = [...new Map(configs.map(config => [config.resolvedHash, config])).values()].map((config) => { + const heartRate = config.label === 'default' ? heartRates.default : heartRates.candidate + const readings = Array.from({ length: 10 }, (_, index) => ({ side: 'left' as const, timestampMs: night.startMs + index * 60000, heartRate, hrvIndex: 30, + hrvMetric: 'window_ibi_variability_v1' as const, breathingRate: 14, qualityScore: 1, flags: null, hrRaw: heartRate, sourceSide: 'left' as const, + configHash: config.resolvedHash, limitations: [] })) + return { cohortId: night.cohortId, configHash: config.resolvedHash, readings, run: { version: 'retained-waveform-run-v1' as const, build_hash: pythonBuildHash, + decoded_records: 10, raw_records: 10, counts: { records: 10, accepted_frames: 10, empty_channel_frames: 0, pump_gated_frames: 0, preroll_outputs: 0, scored_outputs: readings.length }, + limitations: [], config_hashes: { left: config.resolvedHash, right: defaults.resolvedHash }, artifact_hashes: [artifact] } } + }) + const protocol = { ...DEFAULT_PROTOCOL } + const job = { version: 'backtest-job-v1' as const, experimentId: '00000000-0000-4000-8000-000000000000', idempotencyKey: `local:${id}:${split}`, evaluatorBuildHash: 'e'.repeat(64), + split, configs, waveform: { pythonBuildHash, otherSide: defaults }, datasetId: fixture.id, datasetHash: fixture.contentHash, + protocol, protocolHash: configHash(protocol), holdoutUse: 0, evidence: 'descriptive_only' as const } + const report = evaluateBacktest({ job, dataset: { manifest: fixture.manifest, snapshot: fixture.snapshot }, waveform: { hash: dataHash(outputs), outputs } }) + return summarizeLocalReport({ candidate: { id, overrides, resolvedHash: candidate.hash }, split, report, dataset: { id: fixture.id, contentHash: fixture.contentHash }, + build: { pythonBuildHash, pythonSourcesHash: 'c'.repeat(64), evaluatorBuildHash: 'e'.repeat(64) }, protocol }) +} + +describe('local fit reports', () => { + it('labels every report exploratory-local and scores HR, breathing and HRV per night', () => { + const report = localReport('tighter', [{ path: 'heart_rate.trackingToleranceBpm', value: 10 }], 'tuning', { default: 60, candidate: 61 }) + expect(report).toMatchObject({ version: 'biometrics-local-fit-report-v1', label: 'exploratory-local', evidence: 'not_activation_evidence', split: 'tuning', + candidate: { id: 'tighter', resolvedHash: resolveConfig([{ path: 'heart_rate.trackingToleranceBpm', value: 10 }]).hash } }) + expect(report.notice).toContain('Only a backtest run on the pod') + expect(report.summary.heart_rate).toMatchObject({ status: 'scored', totalNights: 1, scoredNights: 1, coverageReducedNights: 0, + equalNightMeanMae: { default: 2, candidate: 1 }, equalNightMeanMaeDifference: -1 }) + expect(report.summary.heart_rate.nights).toEqual([{ cohortId: 'night-a', pairedCount: 5, defaultMae: 2, candidateMae: 1, maeDifference: -1, coverageReduced: false, lostObservationTimestamps: 0 }]) + // The fixture has no breathing or HRV reference, which must read as unscored, never zero error. + expect(report.summary.breathing_rate).toMatchObject({ status: 'no_paired_reference', scoredNights: 0, equalNightMeanMaeDifference: null }) + expect(report.summary.window_ibi_variability_v1.status).toBe('no_paired_reference') + expect(report.report.status).toBe('insufficient_evidence') + expect(parseLocalReport(JSON.parse(JSON.stringify(report))).label).toBe('exploratory-local') + expect(() => parseLocalReport({ ...report, label: 'pod' })).toThrow() + }) + + it('ranks by tuning HR error difference and keeps validation alongside', () => { + const better = [{ path: 'heart_rate.trackingToleranceBpm', value: 10 }], worse = [{ path: 'heart_rate.trackingToleranceBpm', value: 25 }] + const reports = [ + localReport('worse', worse, 'tuning', { default: 60, candidate: 58 }), localReport('worse', worse, 'validation', { default: 60, candidate: 59 }), + localReport('better', better, 'tuning', { default: 60, candidate: 61.5 }), localReport('better', better, 'validation', { default: 60, candidate: 60 }), + localReport('defaults', [], 'tuning', { default: 60, candidate: 60 }), localReport('defaults', [], 'validation', { default: 60, candidate: 60 }), + ] + const ranking = rankCandidates(reports) + expect(ranking).toMatchObject({ label: 'exploratory-local', evidence: 'not_activation_evidence' }) + expect(ranking.rows.map(row => [row.rank, row.id])).toEqual([[1, 'better'], [2, 'defaults'], [3, 'worse']]) + expect(ranking.rows[0].heartRate).toEqual({ tuning: { maeDifference: -1.5, mae: 0.5, scoredNights: 1, coverageReducedNights: 0 }, + validation: { maeDifference: 0, mae: 2, scoredNights: 1, coverageReducedNights: 0 } }) + const tied = rankCandidates([localReport('b-two', [...better, { path: 'heart_rate.trackingHistoryLength', value: 8 }], 'tuning', { default: 60, candidate: 61 }), + localReport('a-one', [...better, { path: 'hrv_index.minimumWindows', value: 10 }], 'tuning', { default: 60, candidate: 61 })]) + expect(tied.rows.map(row => [row.id, row.changedDefaults])).toEqual([['a-one', 1], ['b-two', 2]]) + expect(rankingMarkdown(ranking)).toContain('| 1 | better | 0.500 | -1.500 | 2.000 | 0.000 | 0/0 | n/a | n/a |') + }) + + it('compares local default HR error with pod results within 0.1 bpm', () => { + const defaults = [localReport('defaults', [], 'tuning', { default: 60, candidate: 60 }), localReport('defaults', [], 'validation', { default: 60, candidate: 60 })] + const pod = (split: string, mae: number) => ({ report: { split, datasetHash: fixture.contentHash, numericNightSummary: [{ metric: 'heart_rate', baseline: 'default', candidate: 'current', + baselineHash: configHash(defaultConfig), candidateHash: configHash(defaultConfig), + nights: [{ cohortId: split === 'tuning' ? 'night-a' : 'night-b', commonSupport: { baseline: { pairedCount: 5, mae } } }] }] } }) + const close = compareWithPod(defaults, [pod('tuning', 2.05), pod('validation', 2)]) + expect(close.allWithinTolerance).toBe(true) + expect(close.rows.map(row => [row.cohortId, row.localMae, row.podMae])).toEqual([['night-a', 2, 2.05], ['night-b', 2, 2]]) + expect(compareWithPod(defaults, [pod('tuning', 2.2)]).allWithinTolerance).toBe(false) + expect(() => compareWithPod(defaults, [pod('holdout', 2)])).toThrow('holdout is never replayed locally') + expect(() => compareWithPod(defaults, [{ report: { ...pod('tuning', 2).report, datasetHash: 'f'.repeat(64) } }])).toThrow('different dataset') + }) +}) + +describe('local fit inputs', () => { + it('never replays holdout nights', async () => { + const bundle = { index: { dataset: { id: fixture.id, contentHash: fixture.contentHash } }, manifest: fixture.manifest, snapshot: fixture.snapshot, rawDirectory: fixture.rawDirectory } + await expect(fitCandidates({ bundle: bundle as never, candidates: [{ id: 'x', overrides: [] }], outputDirectory: directory, splits: ['holdout' as never] })) + .rejects.toThrow('Holdout runs once, on the pod') + }) + + it('expands grids, including jointly varied axes', () => { + const candidates = expandGrid(DEFAULT_GRID) + expect(candidates).toHaveLength(18) + expect(new Set(candidates.map(row => row.id)).size).toBe(18) + expect(candidates[0]).toEqual({ id: 'tol10-hist3-brlo0.5-brhi1.5-mw10', overrides: [ + { path: 'heart_rate.trackingToleranceBpm', value: 10 }, { path: 'heart_rate.trackingHistoryLength', value: 3 }, + { path: 'breathing_rate.intervalLowerRatio', value: 0.5 }, { path: 'breathing_rate.intervalUpperRatio', value: 1.5 }, { path: 'hrv_index.minimumWindows', value: 10 }] }) + for (const candidate of candidates) expect(() => resolveConfig(candidate.overrides as never)).not.toThrow() + expect(() => expandGrid({ 'a,b': [[1]] })).toThrow('needs 2 values') + }) +}) diff --git a/scripts/biometrics-local-fit/report.ts b/scripts/biometrics-local-fit/report.ts new file mode 100644 index 00000000..98a87754 --- /dev/null +++ b/scripts/biometrics-local-fit/report.ts @@ -0,0 +1,120 @@ +import { z } from 'zod' +import type { evaluateBacktest } from '@/src/biometrics/backtest/evaluator' +import { defaultConfig, type Override } from '@/src/biometrics/algorithms/registry' +import type { createNumericNightSummary } from '@/src/biometrics/backtest/numeric-night-summary' + +export const METRICS = ['heart_rate', 'breathing_rate', 'window_ibi_variability_v1'] as const +type Metric = typeof METRICS[number] +type EvaluatorReport = ReturnType +type NightSummary = ReturnType>[number] + +const mean = (values: number[]) => values.length ? values.reduce((sum, value) => sum + value, 0) / values.length : null + +function metricSummary(report: EvaluatorReport, metric: Metric) { + const summary = (report.numericNightSummary as NightSummary[] | undefined) + ?.find(row => row.metric === metric && row.baseline === 'default' && row.candidate === 'candidate') + const nights = (summary?.nights ?? []).map(night => ({ cohortId: night.cohortId, pairedCount: night.commonSupport.candidate.pairedCount, + defaultMae: night.commonSupport.baseline.mae, candidateMae: night.commonSupport.candidate.mae, maeDifference: night.commonMaeDifference, + coverageReduced: night.coverageReduced, lostObservationTimestamps: night.lostObservationTimestamps })) + const scored = nights.filter(night => night.maeDifference !== null) + return { status: scored.length ? 'scored' as const : 'no_paired_reference' as const, + totalNights: nights.length, scoredNights: scored.length, coverageReducedNights: nights.filter(night => night.coverageReduced).length, + equalNightMeanMae: { default: mean(scored.map(night => night.defaultMae as number)), candidate: mean(scored.map(night => night.candidateMae as number)) }, + equalNightMeanMaeDifference: summary?.equalNightMeanMaeDifference ?? null, nights } +} +export type MetricSummary = ReturnType + +/** One report per candidate per split. Everything here was computed off-pod + * and is labelled so it cannot be mistaken for the pod's backtest evidence. */ +export function summarizeLocalReport(input: { + candidate: { id: string, overrides: Array<{ path: string, value: unknown }>, resolvedHash: string } + split: 'tuning' | 'validation' + report: EvaluatorReport + dataset: { id: string, contentHash: string } + build: { pythonBuildHash: string, pythonSourcesHash: string, evaluatorBuildHash: string } + protocol: unknown +}) { + return { version: 'biometrics-local-fit-report-v1' as const, label: 'exploratory-local' as const, + evidence: 'not_activation_evidence' as const, + notice: 'Computed off-pod from an exported dataset bundle. Only a backtest run on the pod can support activation; create the candidate with propose_biometric_profile and backtest it there.', + candidate: { id: input.candidate.id, overrides: input.candidate.overrides as Override[], resolvedHash: input.candidate.resolvedHash }, + split: input.split, dataset: { id: input.dataset.id, contentHash: input.dataset.contentHash }, build: input.build, protocol: input.protocol, + summary: Object.fromEntries(METRICS.map(metric => [metric, metricSummary(input.report, metric)])) as Record, + report: input.report } +} +export type LocalFitReport = ReturnType + +const reportShape = z.object({ version: z.literal('biometrics-local-fit-report-v1'), label: z.literal('exploratory-local'), + evidence: z.literal('not_activation_evidence'), candidate: z.object({ id: z.string(), resolvedHash: z.string() }).passthrough(), + split: z.enum(['tuning', 'validation']), summary: z.record(z.string(), z.object({ status: z.string(), nights: z.array(z.unknown()) }).passthrough()) }).passthrough() +export function parseLocalReport(value: unknown) { + return reportShape.parse(value) as unknown as LocalFitReport +} + +/** Rank by the tuning split's equal-night mean HR error difference against + * defaults on common support (lower is better), validation shown alongside. */ +export function rankCandidates(reports: LocalFitReport[]) { + const ids = [...new Set(reports.map(report => report.candidate.id))] + const pick = (id: string, split: 'tuning' | 'validation') => reports.find(report => report.candidate.id === id && report.split === split) + const view = (report: LocalFitReport | undefined, metric: Metric) => report + ? { maeDifference: report.summary[metric].equalNightMeanMaeDifference, mae: report.summary[metric].equalNightMeanMae.candidate, + scoredNights: report.summary[metric].scoredNights, coverageReducedNights: report.summary[metric].coverageReducedNights } + : null + const key = (value: number | null | undefined) => value ?? Number.POSITIVE_INFINITY + // Ties go to the candidate that changes the fewest defaults. + const changes = (overrides: Override[]) => overrides.filter(row => row.path.split('.') + .reduce((parent, part) => (parent as Record | undefined)?.[part], defaultConfig) !== row.value).length + const rows = ids.map((id) => { + const tuning = pick(id, 'tuning'), validation = pick(id, 'validation') + const candidate = (tuning ?? validation)?.candidate + return { id, overrides: candidate?.overrides ?? [], resolvedHash: candidate?.resolvedHash ?? '', changedDefaults: changes(candidate?.overrides ?? []), + heartRate: { tuning: view(tuning, 'heart_rate'), validation: view(validation, 'heart_rate') }, + breathingRate: { tuning: view(tuning, 'breathing_rate'), validation: view(validation, 'breathing_rate') }, + hrv: { tuning: view(tuning, 'window_ibi_variability_v1'), validation: view(validation, 'window_ibi_variability_v1') } } + }) + rows.sort((a, b) => key(a.heartRate.tuning?.maeDifference) - key(b.heartRate.tuning?.maeDifference) + || key(a.heartRate.validation?.maeDifference) - key(b.heartRate.validation?.maeDifference) + || (a.heartRate.tuning?.coverageReducedNights ?? 0) - (b.heartRate.tuning?.coverageReducedNights ?? 0) + || a.changedDefaults - b.changedDefaults + || a.id.localeCompare(b.id)) + return { version: 'biometrics-local-fit-ranking-v1' as const, label: 'exploratory-local' as const, evidence: 'not_activation_evidence' as const, + rankedBy: 'tuning.heart_rate.equalNightMeanMaeDifference_vs_defaults_common_support' as const, + rows: rows.map((row, index) => ({ rank: index + 1, ...row })) } +} + +const format = (value: number | null | undefined, digits = 3) => value === null || value === undefined ? 'n/a' : value.toFixed(digits) +export function rankingMarkdown(ranking: ReturnType) { + const lines = ['_Exploratory, computed off-pod. Not activation evidence: confirm on the pod._', '', + '| # | candidate | tuning HR MAE | tuning Δ vs defaults | validation HR MAE | validation Δ | coverage-reduced nights (t/v) | BR Δ (t) | HRV Δ (t) |', + '| ---: | --- | ---: | ---: | ---: | ---: | :---: | ---: | ---: |'] + for (const row of ranking.rows) { + lines.push(`| ${row.rank} | ${row.id} | ${format(row.heartRate.tuning?.mae)} | ${format(row.heartRate.tuning?.maeDifference)} | ${format(row.heartRate.validation?.mae)} | ${format(row.heartRate.validation?.maeDifference)} | ${row.heartRate.tuning?.coverageReducedNights ?? 'n/a'}/${row.heartRate.validation?.coverageReducedNights ?? 'n/a'} | ${format(row.breathingRate.tuning?.maeDifference)} | ${format(row.hrv.tuning?.maeDifference)} |`) + } + return `${lines.join('\n')}\n` +} + +const podResultSchema = z.object({ report: z.object({ split: z.enum(['tuning', 'validation', 'holdout']), datasetHash: z.string(), + numericNightSummary: z.array(z.object({ metric: z.string(), baseline: z.string(), candidate: z.string(), baselineHash: z.string(), candidateHash: z.string(), + nights: z.array(z.object({ cohortId: z.string(), commonSupport: z.object({ baseline: z.object({ pairedCount: z.number(), mae: z.number().nullable() }).passthrough() }).passthrough() }).passthrough()) }).passthrough()) }).passthrough() }).passthrough() + +/** Compare local per-night HR error for defaults with the pod's own result + * files (GET /api/biometric-tuning/backtests/{id}/result). */ +export function compareWithPod(defaults: LocalFitReport[], podResults: unknown[], toleranceBpm = 0.1) { + const rows = podResults.flatMap((raw) => { + const report = podResultSchema.parse(raw).report + if (report.split === 'holdout') throw new Error('Parity uses tuning and validation results only; holdout is never replayed locally') + const local = defaults.find(row => row.split === report.split) + const defaultsHash = defaults[0]?.candidate.resolvedHash + const pod = report.numericNightSummary.find(row => row.metric === 'heart_rate' && row.baseline === 'default' && row.baselineHash === defaultsHash) + if (!pod) throw new Error(`Pod ${report.split} result has no heart-rate summary for the shipped defaults`) + if (local && local.dataset.contentHash !== report.datasetHash) throw new Error('Pod result is for a different dataset') + return pod.nights.map((night) => { + const localNight = local?.summary.heart_rate.nights.find(row => row.cohortId === night.cohortId) + const podMae = night.commonSupport.baseline.mae, localMae = localNight?.defaultMae ?? null + const difference = podMae === null || localMae === null ? null : localMae - podMae + return { split: report.split, cohortId: night.cohortId, podMae, localMae, podPairedCount: night.commonSupport.baseline.pairedCount, + localPairedCount: localNight?.pairedCount ?? null, difference, withinTolerance: difference !== null && Math.abs(difference) <= toleranceBpm } + }) + }) + return { version: 'biometrics-local-fit-parity-v1' as const, toleranceBpm, rows, allWithinTolerance: rows.length > 0 && rows.every(row => row.withinTolerance) } +} diff --git a/src/biometrics/backtest/bundle-tar.ts b/src/biometrics/backtest/bundle-tar.ts new file mode 100644 index 00000000..7851f09c --- /dev/null +++ b/src/biometrics/backtest/bundle-tar.ts @@ -0,0 +1,59 @@ +/** Minimal ustar framing for calibration bundles: regular files with short, + * relative names only. Writer and reader live together so their limits agree. */ +const BLOCK = 512 +const NAME = /^[A-Za-z0-9_.-]+(\/[A-Za-z0-9_.-]+)?$/ +export const TAR_END = Buffer.alloc(BLOCK * 2) + +function field(header: Buffer, offset: number, length: number, value: string) { + header.write(value, offset, Math.min(length, Buffer.byteLength(value)), 'ascii') +} + +function octal(value: number, digits: number) { + return value.toString(8).padStart(digits, '0') +} + +export function tarHeader(name: string, bytes: number, mtimeSeconds = 0) { + if (!NAME.test(name) || name.length > 100 || name.split('/').some(part => part === '.' || part === '..')) throw new Error('Invalid bundle entry name') + if (!Number.isSafeInteger(bytes) || bytes < 0 || bytes >= 8 ** 11) throw new Error('Invalid bundle entry size') + const header = Buffer.alloc(BLOCK) + field(header, 0, 100, name) + field(header, 100, 8, '0000600\0') + field(header, 108, 8, '0000000\0') + field(header, 116, 8, '0000000\0') + field(header, 124, 12, `${octal(bytes, 11)}\0`) + field(header, 136, 12, `${octal(Math.max(0, Math.floor(mtimeSeconds)), 11)}\0`) + field(header, 148, 8, ' ') + field(header, 156, 1, '0') + field(header, 257, 6, 'ustar\0') + field(header, 263, 2, '00') + let sum = 0 + for (const byte of header) sum += byte + field(header, 148, 8, `${octal(sum, 6)}\0 `) + return header +} + +export function tarPadding(bytes: number) { + return Buffer.alloc((BLOCK - (bytes % BLOCK)) % BLOCK) +} + +export interface TarEntryHeader { name: string, bytes: number } + +/** Parse one header block. Returns null for the zero end-of-archive block and + * rejects anything other than a checksummed regular file with a safe name. */ +export function parseTarHeader(block: Buffer): TarEntryHeader | null { + if (block.length !== BLOCK) throw new Error('Truncated bundle archive') + if (block.every(byte => byte === 0)) return null + const text = (offset: number, length: number) => block.subarray(offset, offset + length).toString('ascii').replace(/\0.*$/s, '') + let sum = 0 + for (let index = 0; index < BLOCK; index++) sum += index >= 148 && index < 156 ? 32 : block[index] + if (Number.parseInt(text(148, 8).trim(), 8) !== sum) throw new Error('Bundle archive header checksum mismatch') + if (!text(257, 6).startsWith('ustar')) throw new Error('Bundle archive is not ustar') + const type = text(156, 1) + if (type !== '0' && type !== '') throw new Error('Bundle archive may contain only regular files') + if (text(345, 155)) throw new Error('Bundle archive entry names must be short') + const name = text(0, 100) + if (!NAME.test(name) || name.split('/').some(part => part === '.' || part === '..')) throw new Error('Unsafe bundle archive entry name') + const bytes = Number.parseInt(text(124, 12).trim(), 8) + if (!Number.isSafeInteger(bytes) || bytes < 0) throw new Error('Invalid bundle archive entry size') + return { name, bytes } +} diff --git a/src/biometrics/backtest/coordinator-lanes.ts b/src/biometrics/backtest/coordinator-lanes.ts index 14477eec..f9cd7745 100644 --- a/src/biometrics/backtest/coordinator-lanes.ts +++ b/src/biometrics/backtest/coordinator-lanes.ts @@ -2,7 +2,7 @@ import type { CoordinatorTask } from './coordinator-tasks' export type CoordinatorLane = 'read' | 'work' export const coordinatorTaskLanes: Record = { readShadowReport: 'read', - pageRawSources: 'read', datasetDescriptor: 'read', readResult: 'read', readSummary: 'read', readTrace: 'read', readNumericTrace: 'read', readReferenceTrace: 'read', + pageRawSources: 'read', datasetDescriptor: 'read', exportDataset: 'read', readResult: 'read', readSummary: 'read', readTrace: 'read', readNumericTrace: 'read', readReferenceTrace: 'read', createWaveformDataset: 'work', inspectRawSource: 'work', createDataset: 'work', releaseDataset: 'work', releaseResult: 'work', createExperiment: 'work', collectArtifacts: 'work', execute: 'work', } diff --git a/src/biometrics/backtest/coordinator-tasks.ts b/src/biometrics/backtest/coordinator-tasks.ts index bcc8e614..9ab1f299 100644 --- a/src/biometrics/backtest/coordinator-tasks.ts +++ b/src/biometrics/backtest/coordinator-tasks.ts @@ -12,6 +12,7 @@ import { summarizeBacktestResult } from './report-summary' import type { BetterSQLite3Database } from 'drizzle-orm/better-sqlite3' import type * as schema from '@/src/db/biometrics-schema' import { DatasetStore, createDatasetSchema } from './datasets' +import { prepareDatasetExport } from './dataset-export' import { BacktestJobStore, experimentProtocolSchema } from './jobs' import { readStageTraces, readNumericTraces, readReferenceTraces } from './traces' import { traceRequestSchema } from './trace-contracts' @@ -31,6 +32,7 @@ export const coordinatorTaskSchema = z.discriminatedUnion('type', [ z.object({ type: z.literal('pageRawSources'), input: rawSourcePageSchema }).strict(), z.object({ type: z.literal('createDataset'), input: createDatasetSchema }).strict(), z.object({ type: z.literal('datasetDescriptor'), input: idInput }).strict(), + z.object({ type: z.literal('exportDataset'), input: idInput }).strict(), z.object({ type: z.literal('readResult'), input: idInput }).strict(), z.object({ type: z.literal('readSummary'), input: idInput }).strict(), z.object({ type: z.literal('readReferenceTrace'), input: referenceTraceRequestSchema }).strict(), @@ -50,6 +52,7 @@ export interface CoordinatorResults { pageRawSources: ReturnType createDataset: ReturnType datasetDescriptor: ReturnType + exportDataset: Awaited> readResult: ReturnType readSummary: ReturnType>>> readReferenceTrace: ReturnType @@ -79,6 +82,7 @@ export async function performCoordinatorTask(db: BetterSQLite3Database file.path === `raw/${file.sha256}.RAW.gz`, 'Raw bundle path must name its content hash'), +]) +/** The bundle's own table of contents. Every other file is listed with its exact + * bytes and SHA-256; a reader must reject anything unlisted. */ +export const datasetBundleIndexSchema = z.object({ + version: z.literal('calibration-dataset-bundle-v1'), + dataset: z.object({ id: z.string().uuid(), side: z.enum(['left', 'right']), subjectId: z.string(), contentHash: hash, + snapshotVersion: z.enum(['processed-dataset-v1', 'waveform-dataset-v1']), createdAtMs: size }).strict(), + files: z.array(datasetBundleFileSchema).min(2).max(1026), + limitations: z.array(z.string().max(500)).max(10), +}).strict() +export type DatasetBundleIndex = z.infer + +export const DATASET_BUNDLE_INDEX_PATH = 'bundle.json' +const LIMITATIONS = [ + 'Contains only this frozen dataset: its manifest, snapshot and the retained RAW archives it references.', + 'Off-pod replay of this bundle is exploratory; only a backtest on the pod can serve as activation evidence.', + 'Dual-channel RAW archives include the other side of the bed.', +] + +export function retainedRawDirectory(db: BetterSQLite3Database) { + const path = db.all<{ name: string, file: string }>(sql`PRAGMA database_list`).find(row => row.name === 'main')?.file + if (!path) throw new ReferenceInputError('WORKER_UNAVAILABLE', 'Dataset export requires a persistent biometrics database') + return join(dirname(path), 'biometric-raw') +} + +/** Open a retained artifact without following links; size must match its record. */ +export async function openRetainedArtifact(directory: string, name: string, expectedBytes: number) { + if (!/^[a-f0-9]{64}\.RAW\.gz$/.test(name)) throw new ReferenceInputError('DATASET_CORRUPT', 'Invalid retained artifact name') + const handle = await open(join(directory, name), constants.O_RDONLY | constants.O_NOFOLLOW).catch(() => { + throw new ReferenceInputError('DATASET_CORRUPT', 'A retained RAW archive is missing') + }) + const info = await handle.stat() + if (!info.isFile() || info.size !== expectedBytes) { + await handle.close() + throw new ReferenceInputError('DATASET_CORRUPT', 'A retained RAW archive changed size') + } + return handle +} + +async function hashRetainedArtifact(directory: string, name: string, expectedBytes: number) { + const handle = await openRetainedArtifact(directory, name, expectedBytes) + const digest = createHash('sha256') + for await (const chunk of createReadStream('', { fd: handle.fd, autoClose: false, highWaterMark: 1024 * 1024 })) digest.update(chunk as Buffer) + await handle.close() + return digest.digest('hex') +} + +/** Verify a live dataset and every retained archive it references before anything + * is streamed. Released or missing datasets surface as NOT_FOUND/DATASET_RELEASED. */ +export async function prepareDatasetExport(db: BetterSQLite3Database, id: string, retainedDirectory = retainedRawDirectory(db)) { + const dataset = new DatasetStore(db).read(id) + const references = db.select({ position: schema.biometricDatasetRawArtifacts.position, hash: schema.biometricRawArtifacts.contentHash, + name: schema.biometricRawArtifacts.name, compressedBytes: schema.biometricRawArtifacts.compressedBytes, expandedBytes: schema.biometricRawArtifacts.expandedBytes }) + .from(schema.biometricDatasetRawArtifacts) + .innerJoin(schema.biometricRawArtifacts, eq(schema.biometricRawArtifacts.contentHash, schema.biometricDatasetRawArtifacts.artifactHash)) + .where(and(eq(schema.biometricDatasetRawArtifacts.datasetId, id), isNull(schema.biometricDatasetRawArtifacts.releasedAtMs))) + .orderBy(asc(schema.biometricDatasetRawArtifacts.position)).all() + const frozen = new Map(dataset.snapshot.version === 'waveform-dataset-v1' + ? dataset.snapshot.waveformInputs.flatMap(cohort => cohort.input.artifacts.map(source => [source.artifact.sha256, source.artifact] as const)) + : []) + if (references.length !== frozen.size || references.some((row) => { + const artifact = frozen.get(row.hash) + return !artifact || row.name !== artifact.name || row.compressedBytes !== artifact.compressed_bytes || row.expandedBytes !== artifact.expanded_bytes + })) throw new ReferenceInputError('DATASET_CORRUPT', 'Retained RAW references do not match the frozen dataset') + const raw = [] + for (const row of references) { + if (await hashRetainedArtifact(retainedDirectory, row.name, row.compressedBytes) !== row.hash) { + throw new ReferenceInputError('DATASET_CORRUPT', 'A retained RAW archive failed its SHA-256 check') + } + raw.push({ role: 'raw' as const, path: `raw/${row.name}`, sha256: row.hash, bytes: row.compressedBytes, expandedBytes: row.expandedBytes }) + } + const manifestText = JSON.stringify(dataset.manifest) + const snapshotText = JSON.stringify(dataset.snapshot) + const text = (role: 'manifest' | 'snapshot', path: 'manifest.json' | 'snapshot.json', value: string) => ({ role, path, + sha256: createHash('sha256').update(value).digest('hex'), bytes: Buffer.byteLength(value) }) + const index = datasetBundleIndexSchema.parse({ version: 'calibration-dataset-bundle-v1', + dataset: { id: dataset.id, side: dataset.side, subjectId: dataset.subjectId, contentHash: dataset.contentHash, + snapshotVersion: dataset.snapshot.version, createdAtMs: dataset.createdAtMs }, + files: [text('manifest', 'manifest.json', manifestText), text('snapshot', 'snapshot.json', snapshotText), ...raw], + limitations: LIMITATIONS }) + return { index, manifestText, snapshotText } +} + +/** Exact byte length of the uncompressed tar stream the export route writes. */ +export function datasetBundleTarBytes(index: DatasetBundleIndex) { + const indexBytes = Buffer.byteLength(JSON.stringify(index)) + return [indexBytes, ...index.files.map(file => file.bytes)].reduce((sum, bytes) => sum + 512 + Math.ceil(bytes / 512) * 512, 1024) +} + +/** Tar chunks for one prepared export, in index order. */ +export async function* datasetBundleChunks(index: DatasetBundleIndex, texts: { manifest: string, snapshot: string }, directory: string) { + const mtime = index.dataset.createdAtMs / 1000 + const indexText = Buffer.from(JSON.stringify(index)) + yield tarHeader(DATASET_BUNDLE_INDEX_PATH, indexText.length, mtime) + yield indexText + yield tarPadding(indexText.length) + for (const file of index.files) { + if (file.role !== 'raw') { + const body = Buffer.from(file.role === 'manifest' ? texts.manifest : texts.snapshot) + yield tarHeader(file.path, body.length, mtime) + yield body + yield tarPadding(body.length) + continue + } + // Hash again while streaming: verification ran in the coordinator, and the + // bytes sent must be the bytes verified. A mismatch aborts the download. + const handle = await openRetainedArtifact(directory, `${file.sha256}.RAW.gz`, file.bytes) + try { + yield tarHeader(file.path, file.bytes, mtime) + const digest = createHash('sha256') + let sent = 0 + for await (const chunk of createReadStream('', { fd: handle.fd, autoClose: false, highWaterMark: 256 * 1024 })) { + const bytes = chunk as Buffer + digest.update(bytes) + sent += bytes.length + if (sent > file.bytes) throw new Error('Retained archive grew during export') + yield bytes + } + if (sent !== file.bytes || digest.digest('hex') !== file.sha256) throw new Error('Retained archive changed during export') + yield tarPadding(file.bytes) + } + finally { + await handle.close() + } + } + yield TAR_END +} diff --git a/src/biometrics/backtest/tests/dataset-export.test.ts b/src/biometrics/backtest/tests/dataset-export.test.ts new file mode 100644 index 00000000..597bce81 --- /dev/null +++ b/src/biometrics/backtest/tests/dataset-export.test.ts @@ -0,0 +1,146 @@ +// @vitest-environment node +import { createHash } from 'node:crypto' +import { mkdtempSync, rmSync, writeFileSync, readFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { dataHash } from '../../algorithms/canonical' +import { parseTarHeader } from '../bundle-tar' +import { DatasetStore } from '../datasets' +import { datasetBundleChunks, datasetBundleTarBytes, prepareDatasetExport } from '../dataset-export' +import { exportableWaveformDataset } from './fixtures/export' + +const state = vi.hoisted(() => ({ db: undefined as unknown })) +vi.mock('@/src/db', () => ({ + get biometricsDb() { + return state.db + }, +})) +vi.mock('@/src/biometrics/backtest/coordinator-service', () => ({ + coordinatorRequest: (db: Parameters[0], task: { type: string, input: { id: string } }) => { + if (task.type !== 'exportDataset') throw new Error('unexpected task') + return prepareDatasetExport(db, task.input.id) + }, +})) + +let directory: string +let fixture: ReturnType +beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), 'dataset-export-')) + fixture = exportableWaveformDataset(directory) + state.db = fixture.db +}) +afterEach(() => { + fixture.sqlite.close() + rmSync(directory, { recursive: true, force: true }) +}) + +/** Split a tar buffer into its entries using the shared header parser. */ +function untar(buffer: Buffer) { + const entries = new Map() + let position = 0 + for (;;) { + const header = parseTarHeader(buffer.subarray(position, position + 512)) + position += 512 + if (!header) break + entries.set(header.name, buffer.subarray(position, position + header.bytes)) + position += Math.ceil(header.bytes / 512) * 512 + } + return entries +} +const sha256 = (bytes: Buffer) => createHash('sha256').update(bytes).digest('hex') +function tamper(name: string) { + const path = join(fixture.rawDirectory, name) + const bytes = readFileSync(path) + bytes[bytes.length - 1] ^= 0xff + writeFileSync(path, bytes) +} + +describe('prepareDatasetExport', () => { + it('lists the manifest, snapshot and exactly the referenced archives with their SHA-256', async () => { + const { index, manifestText, snapshotText } = await prepareDatasetExport(fixture.db, fixture.id) + expect(index.dataset).toMatchObject({ id: fixture.id, side: 'left', contentHash: fixture.contentHash, snapshotVersion: 'waveform-dataset-v1' }) + expect(index.files.map(file => file.path)).toEqual(['manifest.json', 'snapshot.json', ...fixture.artifacts.map(artifact => `raw/${artifact.name}`)]) + expect(index.files.some(file => file.path.includes(fixture.unrelated.sha256))).toBe(false) + expect(index.files[0]).toMatchObject({ sha256: sha256(Buffer.from(manifestText)), bytes: Buffer.byteLength(manifestText) }) + expect(index.files[1]).toMatchObject({ sha256: sha256(Buffer.from(snapshotText)), bytes: Buffer.byteLength(snapshotText) }) + expect(dataHash({ manifest: JSON.parse(manifestText), snapshot: JSON.parse(snapshotText) })).toBe(fixture.contentHash) + for (const artifact of fixture.artifacts) { + expect(index.files).toContainEqual({ role: 'raw', path: `raw/${artifact.name}`, sha256: artifact.sha256, bytes: artifact.compressed_bytes, expandedBytes: artifact.expanded_bytes }) + } + }) + + it('treats a released dataset as not found', async () => { + new DatasetStore(fixture.db).release(fixture.id) + await expect(prepareDatasetExport(fixture.db, fixture.id)).rejects.toMatchObject({ code: 'DATASET_RELEASED' }) + await expect(prepareDatasetExport(fixture.db, '99999999-9999-4999-8999-999999999999')).rejects.toMatchObject({ code: 'NOT_FOUND' }) + }) + + it('refuses to export an archive whose bytes no longer match its hash', async () => { + tamper(fixture.artifacts[1].name) + await expect(prepareDatasetExport(fixture.db, fixture.id)).rejects.toMatchObject({ code: 'DATASET_CORRUPT', message: expect.stringContaining('SHA-256') }) + }) + + it('refuses missing archives and references that disagree with the frozen snapshot', async () => { + rmSync(join(fixture.rawDirectory, fixture.artifacts[0].name)) + await expect(prepareDatasetExport(fixture.db, fixture.id)).rejects.toMatchObject({ code: 'DATASET_CORRUPT' }) + fixture.sqlite.prepare('UPDATE biometric_dataset_raw_artifacts SET released_at_ms = 1 WHERE artifact_hash = ?').run(fixture.artifacts[0].sha256) + await expect(prepareDatasetExport(fixture.db, fixture.id)).rejects.toMatchObject({ code: 'DATASET_CORRUPT', message: expect.stringContaining('do not match') }) + }) + + it('streams a tar whose size matches the declared length and aborts if an archive changes mid-export', async () => { + const prepared = await prepareDatasetExport(fixture.db, fixture.id) + const chunks: Buffer[] = [] + for await (const chunk of datasetBundleChunks(prepared.index, { manifest: prepared.manifestText, snapshot: prepared.snapshotText }, fixture.rawDirectory)) chunks.push(chunk) + const tar = Buffer.concat(chunks) + expect(tar.length).toBe(datasetBundleTarBytes(prepared.index)) + const entries = untar(tar) + expect([...entries.keys()]).toEqual(['bundle.json', ...prepared.index.files.map(file => file.path)]) + for (const file of prepared.index.files) expect(sha256(entries.get(file.path) as Buffer)).toBe(file.sha256) + + tamper(fixture.artifacts[2].name) + const consume = async () => { + const chunks = datasetBundleChunks(prepared.index, { manifest: prepared.manifestText, snapshot: prepared.snapshotText }, fixture.rawDirectory) + while (!(await chunks.next()).done) { /* drain */ } + } + await expect(consume()).rejects.toThrow('changed during export') + }) +}) + +describe('GET /api/biometric-tuning/datasets/{id}/export', () => { + const get = async (id: string) => { + const { GET } = await import('@/app/api/biometric-tuning/datasets/[id]/export/route') + return GET(new Request(`http://pod/api/biometric-tuning/datasets/${id}/export`), { params: Promise.resolve({ id }) }) + } + + it('streams the verified bundle as an attachment', async () => { + const response = await get(fixture.id) + expect(response.status).toBe(200) + expect(response.headers.get('content-type')).toBe('application/x-tar') + expect(response.headers.get('x-dataset-content-hash')).toBe(fixture.contentHash) + const tar = Buffer.from(await response.arrayBuffer()) + expect(tar.length).toBe(Number(response.headers.get('content-length'))) + const index = JSON.parse((untar(tar).get('bundle.json') as Buffer).toString('utf8')) + expect(index.version).toBe('calibration-dataset-bundle-v1') + expect(untar(tar).size).toBe(1 + 2 + fixture.artifacts.length) + }) + + it('returns 404 for a released or unknown dataset and 400 for a malformed id', async () => { + new DatasetStore(fixture.db).release(fixture.id) + expect((await get(fixture.id)).status).toBe(404) + expect((await get('99999999-9999-4999-8999-999999999999')).status).toBe(404) + expect((await get('not-a-uuid')).status).toBe(400) + }) + + it('returns 409 without streaming when a retained archive fails its hash check', async () => { + tamper(fixture.artifacts[0].name) + const response = await get(fixture.id) + expect(response.status).toBe(409) + expect(await response.json()).toMatchObject({ error: expect.stringContaining('DATASET_CORRUPT') }) + // The failed attempt does not latch the one-at-a-time guard. + tamper(fixture.artifacts[0].name) + const retry = await get(fixture.id) + expect(retry.status).toBe(200) + await retry.arrayBuffer() + }) +}) diff --git a/src/biometrics/backtest/tests/fixtures/export.ts b/src/biometrics/backtest/tests/fixtures/export.ts new file mode 100644 index 00000000..4a2285d3 --- /dev/null +++ b/src/biometrics/backtest/tests/fixtures/export.ts @@ -0,0 +1,63 @@ +import { createHash, randomUUID } from 'node:crypto' +import { mkdirSync, writeFileSync } from 'node:fs' +import { join } from 'node:path' +import { gzipSync } from 'node:zlib' +import Database from 'better-sqlite3' +import { drizzle } from 'drizzle-orm/better-sqlite3' +import { migrate } from 'drizzle-orm/better-sqlite3/migrator' +import * as schema from '@/src/db/biometrics-schema' +import { dataHash } from '../../../algorithms/canonical' +import { previewReferenceImport } from '../../../references/normalize' +import { datasetManifestSchema, datasetSnapshotSchema } from '../../contracts' + +export const exportStartMs = 1791673200000 +const DAY = 86400000, NIGHT = 600000 +export const exportNights = [ + { cohortId: 'night-a', split: 'tuning' as const, startMs: exportStartMs }, + { cohortId: 'night-b', split: 'validation' as const, startMs: exportStartMs + DAY }, + { cohortId: 'night-c', split: 'holdout' as const, startMs: exportStartMs + 2 * DAY }, +] + +/** A complete retained waveform dataset in a file-backed biometrics database: + * one gzip archive per night in DIRECTORY/biometric-raw, plus one unreferenced + * archive that an export must leave out. */ +export function exportableWaveformDataset(directory: string) { + const sqlite = new Database(join(directory, 'biometrics.db')) + sqlite.pragma('foreign_keys = ON') + const db = drizzle(sqlite, { schema }) + migrate(db, { migrationsFolder: 'src/db/biometrics-migrations' }) + const rawDirectory = join(directory, 'biometric-raw') + mkdirSync(rawDirectory, { mode: 0o700 }) + const archive = (text: string) => { + const expanded = Buffer.from(text) + const bytes = gzipSync(expanded) + const sha256 = createHash('sha256').update(bytes).digest('hex') + writeFileSync(join(rawDirectory, `${sha256}.RAW.gz`), bytes, { mode: 0o600 }) + return { name: `${sha256}.RAW.gz`, sha256, compressed_bytes: bytes.length, expanded_bytes: expanded.length } + } + const artifacts = exportNights.map(night => archive(`fixture waveform ${night.cohortId}`)) + const unrelated = archive('another dataset') + const sessions = exportNights.map((night, index) => ({ id: index + 1, startMs: night.startMs, endMs: night.startMs + NIGHT, sleepDurationSeconds: 600, + timesExitedBed: 0, presentIntervals: null, notPresentIntervals: null, movement: [], + vitals: Array.from({ length: 10 }, (_, id) => ({ id: index * 100 + id, timestampMs: night.startMs + id * 60000, heartRate: 60, hrv: 30, breathingRate: 14, processingRunId: null })) })) + const measurements = previewReferenceImport({ side: 'left', subjectId: 'A', source: 'fixture', rows: exportNights.flatMap(night => + Array.from({ length: 5 }, (_, index) => ({ recordKey: `${night.cohortId}-hr-${index}`, metric: 'heart_rate' as const, value: 62, unit: 'bpm', timestampMs: night.startMs + index * 120000 }))) }).measurements + const snapshot = datasetSnapshotSchema.parse({ version: 'waveform-dataset-v1', side: 'left', subjectId: 'A', sessions, processingRuns: [], + references: [{ importId: 'ref', revision: 1, contentHash: 'fixture', source: 'fixture', timePolicy: {}, measurements }], + waveformInputs: exportNights.map((night, index) => ({ cohortId: night.cohortId, input: { version: 'retained-waveform-input-v1', + retained_start_ms: night.startMs, score_start_ms: night.startMs, score_end_ms: night.startMs + NIGHT, + artifacts: [{ artifact: artifacts[index], first_event_ms: night.startMs, last_event_ms: night.startMs + NIGHT - 1 }], contexts: [] } })) }) + const manifest = datasetManifestSchema.parse({ version: 'waveform-dataset-manifest-v1', + splitProtocol: { version: 'chronological-whole-night-v1', timezone: 'UTC', timezoneRuntime: { icu: 'fixture', tz: 'fixture' }, grouping: 'local_noon_connected_sessions', + assignments: exportNights.map((night, index) => ({ cohortId: night.cohortId, nightLabels: [night.cohortId], sessionIds: [index + 1], startMs: night.startMs, endMs: night.startMs + NIGHT, split: night.split })), + evidence: 'partitioned_not_validated', limitations: [] }, + coverage: [], capabilities: { comparison: true, stageReplay: true, signalReplay: true }, limitations: [] }) + const id = randomUUID() + const contentHash = dataHash({ manifest, snapshot }) + db.insert(schema.biometricDatasets).values({ id, side: 'left', subjectId: 'A', idempotencyKey: id, requestHash: id, contentHash, + manifest, snapshot, byteLength: Buffer.byteLength(JSON.stringify({ manifest, snapshot })), createdAtMs: exportStartMs }).run() + for (const artifact of [...artifacts, unrelated]) db.insert(schema.biometricRawArtifacts).values({ contentHash: artifact.sha256, name: artifact.name, + compressedBytes: artifact.compressed_bytes, expandedBytes: artifact.expanded_bytes, createdAtMs: exportStartMs }).run() + artifacts.forEach((artifact, position) => db.insert(schema.biometricDatasetRawArtifacts).values({ id: randomUUID(), datasetId: id, artifactHash: artifact.sha256, position }).run()) + return { sqlite, db, id, rawDirectory, artifacts, unrelated, manifest, snapshot, contentHash } +} diff --git a/src/biometrics/backtest/waveform-child.ts b/src/biometrics/backtest/waveform-child.ts index 337a4c1f..862184e7 100644 --- a/src/biometrics/backtest/waveform-child.ts +++ b/src/biometrics/backtest/waveform-child.ts @@ -14,6 +14,8 @@ export interface WaveformWorkerLocation { pythonPath: string modulesPath: string retainedDirectory: string + /** Off-pod entry script that calls the same replay_worker.run(); unset on the pod. */ + script?: string } /** Locations are trusted server configuration, never fields accepted by RPC. */ @@ -23,7 +25,7 @@ export function runWaveformChild(location: WaveformWorkerLocation, work: unknown expectedBuildHash?: string cpuSeconds?: number } = {}): Promise<{ readings: z.infer[], run: z.infer }> { - if (options.cpuSeconds !== undefined && (!Number.isSafeInteger(options.cpuSeconds) || options.cpuSeconds < 1 || options.cpuSeconds > 3600)) return Promise.reject(new Error('Invalid waveform CPU budget')) + if (options.cpuSeconds !== undefined && (location.script !== undefined || !Number.isSafeInteger(options.cpuSeconds) || options.cpuSeconds < 1 || options.cpuSeconds > 3600)) return Promise.reject(new Error('Invalid waveform CPU budget')) if (options.expectedBuildHash !== undefined) hash.parse(options.expectedBuildHash) const serialized = JSON.stringify(options.expectedBuildHash === undefined ? work : { ...z.record(z.string(), z.unknown()).parse(work), expected_build_hash: options.expectedBuildHash }) if (!serialized || Buffer.byteLength(serialized) > 8 * 1024 * 1024) return Promise.reject(new Error('Waveform input exceeds 8 MiB')) @@ -31,7 +33,8 @@ export function runWaveformChild(location: WaveformWorkerLocation, work: unknown if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 10 || timeoutMs > 3600000) return Promise.reject(new Error('Invalid waveform time budget')) if (options.signal?.aborted) return Promise.reject(new Error('Waveform replay cancelled')) return new Promise((resolve, reject) => { - const child: ChildProcessWithoutNullStreams = spawn(location.pythonPath, ['-B', '-m', 'common.biometrics.replay_worker', '--retained-directory', location.retainedDirectory, ...(options.cpuSeconds === undefined ? [] : ['--cpu-seconds', String(options.cpuSeconds)])], { + const entry = location.script === undefined ? ['-m', 'common.biometrics.replay_worker'] : [location.script] + const child: ChildProcessWithoutNullStreams = spawn(location.pythonPath, ['-B', ...entry, '--retained-directory', location.retainedDirectory, ...(options.cpuSeconds === undefined ? [] : ['--cpu-seconds', String(options.cpuSeconds)])], { cwd: tmpdir(), env: { NODE_ENV: 'production', PYTHONPATH: location.modulesPath, LANG: 'C.UTF-8', TZ: 'UTC', OPENBLAS_NUM_THREADS: '1', OMP_NUM_THREADS: '1', MKL_NUM_THREADS: '1', NUMEXPR_NUM_THREADS: '1' }, stdio: ['pipe', 'pipe', 'pipe'], diff --git a/src/mcp/README.md b/src/mcp/README.md index 693545c5..43cdfd2c 100644 --- a/src/mcp/README.md +++ b/src/mcp/README.md @@ -141,6 +141,7 @@ app/api/mcp/route.ts Next.js route handler (stateless Streamable HTTP) | `import_biometric_reference` | Preview and commit normalized references using the exact preview hash | | `propose_biometric_profile` | Save an immutable agent-origin candidate; never activate | | `manage_biometric_experiment` | Freeze datasets, inspect archive cursors, retain selected waveform sources, and create/list/read experiments with pinned protocols | +| `export_calibration_dataset` | Verify a frozen dataset and its retained archives; return the hashed bundle index and the LAN tar download path (no waveform bytes in the MCP response) | | `backtest_biometrics` | Enqueue/status/list/cancel and compact report summaries | | `apply_biometric_profile` | Explicitly authorized validated/manual activation, reset or rollback; validated activation pins evaluation/report identity, manual activation requires unvalidated acknowledgement | | `release_biometric_artifacts` | Explicit reference deletion or dataset/report release; report release requires its content hash | diff --git a/src/mcp/tests/server.test.ts b/src/mcp/tests/server.test.ts index 40d7d94e..392d4bc7 100644 --- a/src/mcp/tests/server.test.ts +++ b/src/mcp/tests/server.test.ts @@ -48,7 +48,7 @@ const callerMock = vi.hoisted(() => { getReplayAdmission: fn(), listRawSources: fn(), inspectRawSource: fn(), createWaveformDataset: fn(), getManifest: fn(), getProfileOverview: fn(), listSessions: fn(), getSessionCapabilities: fn(), listImports: fn(), getProfileRevision: fn(), listProfileRevisions: fn(), compareReadings: fn(), deleteImport: fn(), releaseDataset: fn(), releaseBacktestResult: fn(), getActivationStatus: fn(), listActivationHistory: fn(), previewImport: fn(), commitImport: fn(), - createProfileRevision: fn(), createDataset: fn(), getDataset: fn(), listDatasets: fn(), createExperiment: fn(), + createProfileRevision: fn(), createDataset: fn(), getDataset: fn(), describeDatasetExport: fn(), listDatasets: fn(), createExperiment: fn(), getExperiment: fn(), listExperiments: fn(), enqueueBacktest: fn(), getBacktest: fn(), cancelBacktest: fn(), listBacktests: fn(), getBacktestSummary: fn(), getBacktestTrace: fn(), getBacktestNumericTrace: fn(), getBacktestReferenceTrace: fn(), activateProfile: fn(), activateValidatedProfile: fn(), rollbackProfile: fn(), resetProfile: fn(), }, @@ -117,6 +117,7 @@ describe('sleepypod MCP server', () => { 'apply_sleep_curve', 'backtest_biometrics', 'diagnose_pod', + 'export_calibration_dataset', 'get_automations', 'get_bed_occupancy', 'get_environment', @@ -1116,6 +1117,12 @@ it('routes every remaining inspection, experiment, backtest and release action t await manage({ action: 'list_experiments', input: { side: 'right' } }) expect(api.listExperiments).toHaveBeenCalledWith({ side: 'right' }) + const download = { method: 'GET', path: `/api/biometric-tuning/datasets/${id}/export`, contentType: 'application/x-tar', bytes: 4096 } + api.describeDatasetExport.mockResolvedValue({ index: { version: 'calibration-dataset-bundle-v1', files: [] }, download }) + expect(await call('export_calibration_dataset', { id })).toMatchObject({ download, next: expect.stringContaining('biometrics:local-fit') }) + expect(api.describeDatasetExport).toHaveBeenCalledWith({ id }) + expect((await client.listTools()).tools.find(tool => tool.name === 'export_calibration_dataset')?.annotations).toMatchObject({ readOnlyHint: true, destructiveHint: false }) + const backtest = (request: Record) => call('backtest_biometrics', { request }) api.getBacktest.mockResolvedValue({ id, state: 'running' }) expect(await backtest({ action: 'status', input: { id } })).toEqual({ id, state: 'running' }) diff --git a/src/mcp/tools/biometrics.ts b/src/mcp/tools/biometrics.ts index dde5e7d9..0f15bc12 100644 --- a/src/mcp/tools/biometrics.ts +++ b/src/mcp/tools/biometrics.ts @@ -144,6 +144,13 @@ export function registerBiometricTools(server: McpServer) { } })) + server.registerTool('export_calibration_dataset', { + title: 'Export a frozen calibration dataset for off-pod fitting', + description: 'Verify one frozen dataset and every retained RAW archive it references, then return the bundle index (each file with its SHA-256) and the LAN download path that streams the tar. Waveform bytes are never placed in this response. Download only to a user-authorized machine: archives include both sides of the bed. Released datasets are not found. Off-pod results are exploratory; confirm candidates with backtest_biometrics on the pod.', + inputSchema: id.shape, annotations: read, + }, input => runTool(async () => jsonResult({ ...await getCaller().biometricTuning.describeDatasetExport(input), + next: 'pnpm biometrics:local-fit fit BUNDLE CANDIDATES OUTPUT_DIR (see scripts/biometrics-local-fit/README.md)' }))) + server.registerTool('backtest_biometrics', { title: 'Start, inspect or cancel a biometric evaluation', description: 'Durable replay never applies a profile or changes hardware. Start pins the candidate and expected binding generation. Report returns compact evidence and a report resource; trace returns at most 500 exact stage intervals and numeric_trace returns at most 500 exact recomputed readings for the dataset side using a pinned report hash and cursor; reference_trace pages frozen source records without submitted payloads or annotations. Descriptive or insufficient evidence is not validated improvement.', diff --git a/src/server/routers/biometricTuning.ts b/src/server/routers/biometricTuning.ts index 0fe9eff3..4d528267 100644 --- a/src/server/routers/biometricTuning.ts +++ b/src/server/routers/biometricTuning.ts @@ -32,6 +32,7 @@ import { BacktestJobStore } from '@/src/biometrics/backtest/jobs' import { describeJob, jobDescriptorSchema, resultDescriptorSchema } from '@/src/biometrics/backtest/descriptors' import { experimentProtocolSchema } from '@/src/biometrics/backtest/contracts' import { datasetDescriptorSchema } from '@/src/biometrics/backtest/contracts' +import { datasetBundleIndexSchema, datasetBundleTarBytes } from '@/src/biometrics/backtest/dataset-export' import { deviceSettings } from '@/src/db/schema' import { ReferenceInputError } from '@/src/biometrics/references/time' import { countUnlinkedLegacyReferences, linkLegacyReferences } from '@/src/biometrics/references/legacy' @@ -287,6 +288,13 @@ export const biometricTuningRouter = router({ .output(datasetDescriptorSchema) .query(({ input }) => references(() => coordinatorRequest(biometricsDb, { type: 'datasetDescriptor', input }).then(result => datasetDescriptorSchema.parse(result)))), + describeDatasetExport: publicProcedure + .meta({ openapi: { method: 'GET', path: '/biometric-tuning/datasets/{id}/export-index', protect: false, tags: ['Biometric tuning'] } }) + .input(importIdSchema) + .output(z.object({ index: datasetBundleIndexSchema, download: z.object({ method: z.literal('GET'), path: z.string(), contentType: z.literal('application/x-tar'), bytes: z.number().int().nonnegative() }).strict() }).strict()) + .query(({ input }) => references(() => coordinatorRequest(biometricsDb, { type: 'exportDataset', input }).then(result => ({ index: result.index, + download: { method: 'GET' as const, path: `/api/biometric-tuning/datasets/${input.id}/export`, contentType: 'application/x-tar' as const, bytes: datasetBundleTarBytes(result.index) } })))), + listDatasets: publicProcedure .meta({ openapi: { method: 'GET', path: '/biometric-tuning/datasets', protect: false, tags: ['Biometric tuning'] } }) .input(z.object({ side: sideSchema, beforeId: z.string().uuid().optional(), beforeCreatedAtMs: z.number().int().nonnegative().optional() }).strict() From 77ab5186e74d01798e2e88fbee1f254550808423 Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sat, 17 Oct 2026 14:41:38 -0700 Subject: [PATCH 2/3] fix(biometrics): map export errors by code so a released or unknown dataset returns 404 on the pod, not 500 --- .../biometric-tuning/datasets/[id]/export/route.ts | 13 +++++++------ .../backtest/tests/dataset-export.test.ts | 13 ++++++++++++- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/app/api/biometric-tuning/datasets/[id]/export/route.ts b/app/api/biometric-tuning/datasets/[id]/export/route.ts index ed659e3e..868161fd 100644 --- a/app/api/biometric-tuning/datasets/[id]/export/route.ts +++ b/app/api/biometric-tuning/datasets/[id]/export/route.ts @@ -4,7 +4,6 @@ import { biometricsDb } from '@/src/db' import { coordinatorRequest } from '@/src/biometrics/backtest/coordinator-service' import type { CoordinatorResults } from '@/src/biometrics/backtest/coordinator-tasks' import { datasetBundleChunks, datasetBundleTarBytes, retainedRawDirectory } from '@/src/biometrics/backtest/dataset-export' -import { ReferenceInputError } from '@/src/biometrics/references/time' // One export at a time: each one reads every retained archive of the dataset // twice (verification, then streaming). A stalled client must not latch it. @@ -15,11 +14,13 @@ const EXPORT_WATCHDOG_MS = (() => { })() function errorResponse(error: unknown) { - if (error instanceof ReferenceInputError) { - if (error.code === 'NOT_FOUND' || error.code === 'DATASET_RELEASED') return NextResponse.json({ error: 'Dataset not found or released' }, { status: 404 }) - if (error.code === 'DATASET_CORRUPT') return NextResponse.json({ error: `DATASET_CORRUPT: ${error.message}` }, { status: 409 }) - if (error.code === 'WORKER_UNAVAILABLE') return NextResponse.json({ error: error.message }, { status: 503 }) - } + // Match on the code, not the class: the coordinator's reply is rebuilt as a + // ReferenceInputError that need not be the same class instance this route + // imports once Next has bundled both. + const code = (error as { code?: unknown } | null)?.code + if (code === 'NOT_FOUND' || code === 'DATASET_RELEASED') return NextResponse.json({ error: 'Dataset not found or released' }, { status: 404 }) + if (code === 'DATASET_CORRUPT') return NextResponse.json({ error: `DATASET_CORRUPT: ${(error as Error).message}` }, { status: 409 }) + if (code === 'WORKER_UNAVAILABLE') return NextResponse.json({ error: (error as Error).message }, { status: 503 }) return NextResponse.json({ error: 'Dataset export failed' }, { status: 500 }) } diff --git a/src/biometrics/backtest/tests/dataset-export.test.ts b/src/biometrics/backtest/tests/dataset-export.test.ts index 597bce81..632579b8 100644 --- a/src/biometrics/backtest/tests/dataset-export.test.ts +++ b/src/biometrics/backtest/tests/dataset-export.test.ts @@ -10,7 +10,7 @@ import { DatasetStore } from '../datasets' import { datasetBundleChunks, datasetBundleTarBytes, prepareDatasetExport } from '../dataset-export' import { exportableWaveformDataset } from './fixtures/export' -const state = vi.hoisted(() => ({ db: undefined as unknown })) +const state = vi.hoisted(() => ({ db: undefined as unknown, failure: null as Error | null })) vi.mock('@/src/db', () => ({ get biometricsDb() { return state.db @@ -19,6 +19,7 @@ vi.mock('@/src/db', () => ({ vi.mock('@/src/biometrics/backtest/coordinator-service', () => ({ coordinatorRequest: (db: Parameters[0], task: { type: string, input: { id: string } }) => { if (task.type !== 'exportDataset') throw new Error('unexpected task') + if (state.failure) return Promise.reject(state.failure) return prepareDatasetExport(db, task.input.id) }, })) @@ -31,6 +32,7 @@ beforeEach(() => { state.db = fixture.db }) afterEach(() => { + state.failure = null fixture.sqlite.close() rmSync(directory, { recursive: true, force: true }) }) @@ -132,6 +134,15 @@ describe('GET /api/biometric-tuning/datasets/{id}/export', () => { expect((await get('not-a-uuid')).status).toBe(400) }) + it('maps coordinator error codes even when the error is not this bundle\'s ReferenceInputError class', async () => { + state.failure = Object.assign(new Error('Retained dataset inputs were explicitly released'), { code: 'DATASET_RELEASED' }) + expect((await get(fixture.id)).status).toBe(404) + state.failure = Object.assign(new Error('A retained RAW archive failed its SHA-256 check'), { code: 'DATASET_CORRUPT' }) + expect((await get(fixture.id)).status).toBe(409) + state.failure = new Error('Biometrics coordinator failed') + expect((await get(fixture.id)).status).toBe(500) + }) + it('returns 409 without streaming when a retained archive fails its hash check', async () => { tamper(fixture.artifacts[0].name) const response = await get(fixture.id) From 4d3359cdc5f4e1c6b02288b31d059c25e636e135 Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sat, 17 Oct 2026 15:42:03 -0700 Subject: [PATCH 3/3] feat(biometrics): let local fit score holdout nights on request so a chosen profile can be checked on fresh nights without the pod --- scripts/biometrics-local-fit/README.md | 7 ++++--- scripts/biometrics-local-fit/cli.ts | 10 ++++++---- scripts/biometrics-local-fit/fit.ts | 7 +++++-- scripts/biometrics-local-fit/report.test.ts | 15 +++++++++++--- scripts/biometrics-local-fit/report.ts | 22 +++++++++++---------- 5 files changed, 39 insertions(+), 22 deletions(-) diff --git a/scripts/biometrics-local-fit/README.md b/scripts/biometrics-local-fit/README.md index c7edc0ae..e61d5bc7 100644 --- a/scripts/biometrics-local-fit/README.md +++ b/scripts/biometrics-local-fit/README.md @@ -58,9 +58,10 @@ on a 10-core laptop for seven frozen nights. The tool verifies every file hash, the dataset content hash and the exact set of archives before any replay. It then replays each tuning and validation night once per distinct configuration, longest nights first, across `--jobs` workers -(default: all cores). Each candidate and split is scored with the pod's own -`evaluateBacktest`, against defaults on common support. Holdout nights are -never replayed locally: holdout runs once, on the pod, for the final candidate. +(default: all cores). Holdout nights are skipped unless you pass +`--include-holdout`. Use it only to score a candidate you have already chosen, +once, on nights it was not fitted to. Each candidate and split is scored with the pod's own +`evaluateBacktest`, against defaults on common support. Outputs: diff --git a/scripts/biometrics-local-fit/cli.ts b/scripts/biometrics-local-fit/cli.ts index 08017583..6c02cfa7 100644 --- a/scripts/biometrics-local-fit/cli.ts +++ b/scripts/biometrics-local-fit/cli.ts @@ -8,7 +8,7 @@ import { compareWithPod, parseLocalReport, rankCandidates, rankingMarkdown } fro const usage = `Usage: verify BUNDLE - fit BUNDLE CANDIDATES.json OUTPUT_DIR [--jobs N] [--splits tuning,validation] [--protocol FILE] [--python PATH] + fit BUNDLE CANDIDATES.json OUTPUT_DIR [--jobs N] [--splits tuning,validation] [--include-holdout] [--protocol FILE] [--python PATH] sweep BUNDLE GRID.json|default OUTPUT_DIR [same options as fit] parity OUTPUT_DIR POD_RESULT.json [POD_RESULT.json ...] BUNDLE is the tar from GET /api/biometric-tuning/datasets/{id}/export, or a directory extracted from it.` @@ -17,7 +17,8 @@ function options(args: string[]) { const values: Record = {} const positional: string[] = [] for (let index = 0; index < args.length; index++) { - if (args[index].startsWith('--')) values[args[index].slice(2)] = args[++index] ?? '' + if (args[index] === '--include-holdout') values['include-holdout'] = 'true' + else if (args[index].startsWith('--')) values[args[index].slice(2)] = args[++index] ?? '' else positional.push(args[index]) } return { values, positional } @@ -38,9 +39,10 @@ async function openBundle(path: string, outputDirectory: string) { async function fit(bundlePath: string, candidates: Candidate[], outputDirectory: string, flags: Record) { await mkdir(outputDirectory, { recursive: true }) const bundle = await openBundle(bundlePath, outputDirectory) - const splits = (flags.splits?.split(',') ?? [...LOCAL_SPLITS]) as LocalSplit[] + const includeHoldout = flags['include-holdout'] === 'true' + const splits = (flags.splits?.split(',') ?? [...LOCAL_SPLITS, ...(includeHoldout ? ['holdout'] : [])]) as LocalSplit[] const started = performance.now() - const result = await fitCandidates({ bundle, candidates, outputDirectory, splits, + const result = await fitCandidates({ bundle, candidates, outputDirectory, splits, includeHoldout, jobs: flags.jobs ? Number(flags.jobs) : undefined, protocol: flags.protocol ? await readJson(flags.protocol) as never : undefined, python: flags.python ? defaultPythonLocation(flags.python) : undefined, diff --git a/scripts/biometrics-local-fit/fit.ts b/scripts/biometrics-local-fit/fit.ts index 2238db89..b7d3bf83 100644 --- a/scripts/biometrics-local-fit/fit.ts +++ b/scripts/biometrics-local-fit/fit.ts @@ -15,7 +15,7 @@ import type { VerifiedBundle } from './bundle' import { summarizeLocalReport, type LocalFitReport } from './report' export const LOCAL_SPLITS = ['tuning', 'validation'] as const -export type LocalSplit = typeof LOCAL_SPLITS[number] +export type LocalSplit = typeof LOCAL_SPLITS[number] | 'holdout' /** Matches the protocol of the pod experiment the frozen nights were scored with. */ export const DEFAULT_PROTOCOL = { version: 'descriptive-evaluation-v1', mode: 'signal_replay', assumedStageCalibrationQuality: 0, pointToleranceMs: 30000, clockOffsetMs: 0 } as const const LOCAL_EXPERIMENT_ID = '00000000-0000-4000-8000-000000000000' @@ -94,6 +94,8 @@ export interface FitOptions { candidates: Candidate[] outputDirectory: string splits?: readonly LocalSplit[] + /** Score holdout nights too, once, for a candidate that is already chosen. */ + includeHoldout?: boolean jobs?: number protocol?: z.input python?: PythonLocation @@ -108,7 +110,8 @@ export async function fitCandidates(options: FitOptions) { if (bundle.snapshot.version !== 'waveform-dataset-v1') throw new Error('Local fitting needs a retained waveform dataset') const snapshot = bundle.snapshot const splits = options.splits ?? LOCAL_SPLITS - if (splits.some(split => !LOCAL_SPLITS.includes(split))) throw new Error('Holdout runs once, on the pod, for the final candidate only') + if (splits.some(split => split === 'holdout') && !options.includeHoldout) throw new Error('Holdout nights are held back: pass --include-holdout only to score an already-chosen candidate, once') + if (splits.some(split => split !== 'holdout' && !(LOCAL_SPLITS as readonly string[]).includes(split))) throw new Error('Unknown split') const protocol = experimentProtocolSchema.parse(options.protocol ?? DEFAULT_PROTOCOL) if (protocol.mode !== 'signal_replay') throw new Error('Local fitting scores waveform replay; use a signal_replay protocol') const python = options.python ?? defaultPythonLocation() diff --git a/scripts/biometrics-local-fit/report.test.ts b/scripts/biometrics-local-fit/report.test.ts index 136fd18c..05d3b5f6 100644 --- a/scripts/biometrics-local-fit/report.test.ts +++ b/scripts/biometrics-local-fit/report.test.ts @@ -24,7 +24,7 @@ afterAll(() => { }) /** Score a candidate with fixed synthetic readings (references are 62 bpm). */ -function localReport(id: string, overrides: Array<{ path: string, value: unknown }>, split: 'tuning' | 'validation', heartRates: { default: number, candidate: number }) { +function localReport(id: string, overrides: Array<{ path: string, value: unknown }>, split: 'tuning' | 'validation' | 'holdout', heartRates: { default: number, candidate: number }) { const defaults = { manifestHash, resolvedHash: configHash(defaultConfig), resolved: defaultConfig } const candidate = resolveConfig(overrides as never) const configs = [{ label: 'default' as const, revisionId: null, ...defaults }, @@ -84,6 +84,15 @@ describe('local fit reports', () => { expect(rankingMarkdown(ranking)).toContain('| 1 | better | 0.500 | -1.500 | 2.000 | 0.000 | 0/0 | n/a | n/a |') }) + it('shows holdout error alongside only when holdout nights were scored', () => { + const overrides = [{ path: 'heart_rate.trackingToleranceBpm', value: 10 }] + const without = rankingMarkdown(rankCandidates([localReport('tol10', overrides, 'tuning', { default: 60, candidate: 61 })])) + expect(without).not.toContain('holdout') + const ranking = rankCandidates([localReport('tol10', overrides, 'tuning', { default: 60, candidate: 61 }), localReport('tol10', overrides, 'holdout', { default: 60, candidate: 62 })]) + expect(ranking.rows[0].heartRate.holdout).toEqual({ maeDifference: -2, mae: 0, scoredNights: 1, coverageReducedNights: 0 }) + expect(rankingMarkdown(ranking)).toContain('| 1 | tol10 | 1.000 | -1.000 | n/a | n/a | 0.000 | -2.000 |') + }) + it('compares local default HR error with pod results within 0.1 bpm', () => { const defaults = [localReport('defaults', [], 'tuning', { default: 60, candidate: 60 }), localReport('defaults', [], 'validation', { default: 60, candidate: 60 })] const pod = (split: string, mae: number) => ({ report: { split, datasetHash: fixture.contentHash, numericNightSummary: [{ metric: 'heart_rate', baseline: 'default', candidate: 'current', @@ -99,10 +108,10 @@ describe('local fit reports', () => { }) describe('local fit inputs', () => { - it('never replays holdout nights', async () => { + it('replays holdout nights only when explicitly included', async () => { const bundle = { index: { dataset: { id: fixture.id, contentHash: fixture.contentHash } }, manifest: fixture.manifest, snapshot: fixture.snapshot, rawDirectory: fixture.rawDirectory } await expect(fitCandidates({ bundle: bundle as never, candidates: [{ id: 'x', overrides: [] }], outputDirectory: directory, splits: ['holdout' as never] })) - .rejects.toThrow('Holdout runs once, on the pod') + .rejects.toThrow('Holdout nights are held back') }) it('expands grids, including jointly varied axes', () => { diff --git a/scripts/biometrics-local-fit/report.ts b/scripts/biometrics-local-fit/report.ts index 98a87754..6c4bea53 100644 --- a/scripts/biometrics-local-fit/report.ts +++ b/scripts/biometrics-local-fit/report.ts @@ -28,7 +28,7 @@ export type MetricSummary = ReturnType * and is labelled so it cannot be mistaken for the pod's backtest evidence. */ export function summarizeLocalReport(input: { candidate: { id: string, overrides: Array<{ path: string, value: unknown }>, resolvedHash: string } - split: 'tuning' | 'validation' + split: 'tuning' | 'validation' | 'holdout' report: EvaluatorReport dataset: { id: string, contentHash: string } build: { pythonBuildHash: string, pythonSourcesHash: string, evaluatorBuildHash: string } @@ -46,7 +46,7 @@ export type LocalFitReport = ReturnType const reportShape = z.object({ version: z.literal('biometrics-local-fit-report-v1'), label: z.literal('exploratory-local'), evidence: z.literal('not_activation_evidence'), candidate: z.object({ id: z.string(), resolvedHash: z.string() }).passthrough(), - split: z.enum(['tuning', 'validation']), summary: z.record(z.string(), z.object({ status: z.string(), nights: z.array(z.unknown()) }).passthrough()) }).passthrough() + split: z.enum(['tuning', 'validation', 'holdout']), summary: z.record(z.string(), z.object({ status: z.string(), nights: z.array(z.unknown()) }).passthrough()) }).passthrough() export function parseLocalReport(value: unknown) { return reportShape.parse(value) as unknown as LocalFitReport } @@ -55,7 +55,7 @@ export function parseLocalReport(value: unknown) { * defaults on common support (lower is better), validation shown alongside. */ export function rankCandidates(reports: LocalFitReport[]) { const ids = [...new Set(reports.map(report => report.candidate.id))] - const pick = (id: string, split: 'tuning' | 'validation') => reports.find(report => report.candidate.id === id && report.split === split) + const pick = (id: string, split: 'tuning' | 'validation' | 'holdout') => reports.find(report => report.candidate.id === id && report.split === split) const view = (report: LocalFitReport | undefined, metric: Metric) => report ? { maeDifference: report.summary[metric].equalNightMeanMaeDifference, mae: report.summary[metric].equalNightMeanMae.candidate, scoredNights: report.summary[metric].scoredNights, coverageReducedNights: report.summary[metric].coverageReducedNights } @@ -65,10 +65,10 @@ export function rankCandidates(reports: LocalFitReport[]) { const changes = (overrides: Override[]) => overrides.filter(row => row.path.split('.') .reduce((parent, part) => (parent as Record | undefined)?.[part], defaultConfig) !== row.value).length const rows = ids.map((id) => { - const tuning = pick(id, 'tuning'), validation = pick(id, 'validation') - const candidate = (tuning ?? validation)?.candidate + const tuning = pick(id, 'tuning'), validation = pick(id, 'validation'), holdout = pick(id, 'holdout') + const candidate = (tuning ?? validation ?? holdout)?.candidate return { id, overrides: candidate?.overrides ?? [], resolvedHash: candidate?.resolvedHash ?? '', changedDefaults: changes(candidate?.overrides ?? []), - heartRate: { tuning: view(tuning, 'heart_rate'), validation: view(validation, 'heart_rate') }, + heartRate: { tuning: view(tuning, 'heart_rate'), validation: view(validation, 'heart_rate'), ...(holdout ? { holdout: view(holdout, 'heart_rate') } : {}) }, breathingRate: { tuning: view(tuning, 'breathing_rate'), validation: view(validation, 'breathing_rate') }, hrv: { tuning: view(tuning, 'window_ibi_variability_v1'), validation: view(validation, 'window_ibi_variability_v1') } } }) @@ -84,11 +84,13 @@ export function rankCandidates(reports: LocalFitReport[]) { const format = (value: number | null | undefined, digits = 3) => value === null || value === undefined ? 'n/a' : value.toFixed(digits) export function rankingMarkdown(ranking: ReturnType) { - const lines = ['_Exploratory, computed off-pod. Not activation evidence: confirm on the pod._', '', - '| # | candidate | tuning HR MAE | tuning Δ vs defaults | validation HR MAE | validation Δ | coverage-reduced nights (t/v) | BR Δ (t) | HRV Δ (t) |', - '| ---: | --- | ---: | ---: | ---: | ---: | :---: | ---: | ---: |'] + const holdout = ranking.rows.some(row => row.heartRate.holdout) + const lines = ['_Exploratory, computed off-pod. Not activation evidence._', '', + `| # | candidate | tuning HR MAE | tuning Δ vs defaults | validation HR MAE | validation Δ |${holdout ? ' holdout HR MAE | holdout Δ |' : ''} coverage-reduced nights (t/v) | BR Δ (t) | HRV Δ (t) |`, + `| ---: | --- | ---: | ---: | ---: | ---: |${holdout ? ' ---: | ---: |' : ''} :---: | ---: | ---: |`] for (const row of ranking.rows) { - lines.push(`| ${row.rank} | ${row.id} | ${format(row.heartRate.tuning?.mae)} | ${format(row.heartRate.tuning?.maeDifference)} | ${format(row.heartRate.validation?.mae)} | ${format(row.heartRate.validation?.maeDifference)} | ${row.heartRate.tuning?.coverageReducedNights ?? 'n/a'}/${row.heartRate.validation?.coverageReducedNights ?? 'n/a'} | ${format(row.breathingRate.tuning?.maeDifference)} | ${format(row.hrv.tuning?.maeDifference)} |`) + const held = holdout ? ` ${format(row.heartRate.holdout?.mae)} | ${format(row.heartRate.holdout?.maeDifference)} |` : '' + lines.push(`| ${row.rank} | ${row.id} | ${format(row.heartRate.tuning?.mae)} | ${format(row.heartRate.tuning?.maeDifference)} | ${format(row.heartRate.validation?.mae)} | ${format(row.heartRate.validation?.maeDifference)} |${held} ${row.heartRate.tuning?.coverageReducedNights ?? 'n/a'}/${row.heartRate.validation?.coverageReducedNights ?? 'n/a'} | ${format(row.breathingRate.tuning?.maeDifference)} | ${format(row.hrv.tuning?.maeDifference)} |`) } return `${lines.join('\n')}\n` }