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..868161fd --- /dev/null +++ b/app/api/biometric-tuning/datasets/[id]/export/route.ts @@ -0,0 +1,88 @@ +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' + +// 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) { + // 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 }) +} + +/** + * 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..e61d5bc7 --- /dev/null +++ b/scripts/biometrics-local-fit/README.md @@ -0,0 +1,90 @@ +# 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). 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: + +- `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..6c02cfa7 --- /dev/null +++ b/scripts/biometrics-local-fit/cli.ts @@ -0,0 +1,108 @@ +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] [--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.` + +function options(args: string[]) { + const values: Record = {} + const positional: string[] = [] + for (let index = 0; index < args.length; 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 } +} + +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 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, 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, + 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..b7d3bf83 --- /dev/null +++ b/scripts/biometrics-local-fit/fit.ts @@ -0,0 +1,175 @@ +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] | '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' +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[] + /** Score holdout nights too, once, for a candidate that is already chosen. */ + includeHoldout?: boolean + 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 => 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() + 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..05d3b5f6 --- /dev/null +++ b/scripts/biometrics-local-fit/report.test.ts @@ -0,0 +1,127 @@ +// @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' | '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 }, + { 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('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', + 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('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 nights are held back') + }) + + 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..6c4bea53 --- /dev/null +++ b/scripts/biometrics-local-fit/report.ts @@ -0,0 +1,122 @@ +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' | 'holdout' + 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', '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 +} + +/** 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' | '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 } + : 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'), 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'), ...(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') } } + }) + 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 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) { + 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` +} + +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..632579b8 --- /dev/null +++ b/src/biometrics/backtest/tests/dataset-export.test.ts @@ -0,0 +1,157 @@ +// @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, failure: null as Error | null })) +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') + if (state.failure) return Promise.reject(state.failure) + 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(() => { + state.failure = null + 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('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) + 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()