CR: feat/biometrics-local-fit (sleepypod/core#837)

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/<sha256>.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<typeof setTimeout> | undefined
+  const release = () => {
+    if (watchdog) clearTimeout(watchdog)
+    inflight = false
+  }
+  const body = new ReadableStream<Uint8Array>({
+    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,
+    },
+  })
+}

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/<sha256>.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.

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": {

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/<sha256>.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/<candidate>/<split>.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.

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<typeof exportableWaveformDataset>
+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')
+})