From cad46465114ae194578e7f8940ceb9de75cd47b6 Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sun, 18 Oct 2026 10:48:57 -0700 Subject: [PATCH 1/2] Let off-pod analysis pull a DB snapshot and RAW window over one read-only HTTP call instead of SSH --- .../biometrics/analysis-export/route.test.ts | 236 ++++++++++++++++++ app/api/biometrics/analysis-export/route.ts | 121 +++++++++ docs/biometrics-datasets.md | 56 +++++ src/biometrics/analysis-export-bundle.ts | 177 +++++++++++++ src/biometrics/analysis-export.ts | 93 +++++++ src/biometrics/tests/analysis-export.test.ts | 70 ++++++ src/demo/handlers/biometricTuning.ts | 1 + src/lib/podStorage.ts | 2 +- src/mcp/README.md | 1 + src/mcp/tests/server.test.ts | 12 +- src/mcp/tools/biometrics.ts | 7 + src/server/routers/biometricTuning.ts | 24 ++ 12 files changed, 798 insertions(+), 2 deletions(-) create mode 100644 app/api/biometrics/analysis-export/route.test.ts create mode 100644 app/api/biometrics/analysis-export/route.ts create mode 100644 src/biometrics/analysis-export-bundle.ts create mode 100644 src/biometrics/analysis-export.ts create mode 100644 src/biometrics/tests/analysis-export.test.ts diff --git a/app/api/biometrics/analysis-export/route.test.ts b/app/api/biometrics/analysis-export/route.test.ts new file mode 100644 index 00000000..84170f5a --- /dev/null +++ b/app/api/biometrics/analysis-export/route.test.ts @@ -0,0 +1,236 @@ +// @vitest-environment node +import { createHash, hkdfSync } from 'node:crypto' +import { mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, symlinkSync, truncateSync, unlinkSync, utimesSync, writeFileSync } from 'node:fs' +import { hostname, tmpdir } from 'node:os' +import { join } from 'node:path' +import Database from 'better-sqlite3' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { parseTarHeader } from '@/src/biometrics/backtest/bundle-tar' + +const state = vi.hoisted(() => ({ archive: '', source: '', seed: { source: 'mmc-cid', value: 'cid-0123' } as { source: string, value: string } })) +vi.mock('@/src/lib/podStorage', () => ({ + get ARCHIVE_DIR() { + return state.archive + }, +})) +vi.mock('@/src/db', () => ({ biometricsDb: { $client: { get name() { + return state.source +} } } })) +vi.mock('@/src/homekit/storage', () => ({ readSeed: () => state.seed })) +vi.mock('@/src/hardware/dacMonitor.instance', () => ({ + getDacMonitorIfRunning: () => ({ getLastStatus: () => ({ podVersion: 'J00', sensorLabel: '20600-0003-J55-B0708DE3' }) }), +})) + +const START = Date.UTC(2026, 9, 10, 6) +const END = Date.UTC(2026, 9, 10, 14) +const MINUTE = 60_000 +const journal = JSON.parse(readFileSync(join(process.cwd(), 'src/db/biometrics-migrations/meta/_journal.json'), 'utf-8')) as { entries: Array<{ when: number, tag: string }> } +const latest = journal.entries[journal.entries.length - 1] + +let root: string +let snapshots: string + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), 'analysis-export-test-')) + state.archive = join(root, 'biometrics-archive') + snapshots = join(root, 'tmpfs') + mkdirSync(state.archive) + mkdirSync(snapshots) + vi.stubEnv('TMPDIR', snapshots) + state.source = join(root, 'biometrics.db') + const database = new Database(state.source) + database.pragma('journal_mode = WAL') + database.exec('CREATE TABLE __drizzle_migrations (id INTEGER PRIMARY KEY, hash TEXT NOT NULL, created_at NUMERIC); CREATE TABLE vitals (id INTEGER PRIMARY KEY, heart_rate REAL)') + database.prepare('INSERT INTO __drizzle_migrations (hash, created_at) VALUES (?, ?), (?, ?)').run('old-hash', 1, 'latest-hash', latest.when) + // Left in the WAL (no checkpoint) so the snapshot must include uncheckpointed pages. + database.prepare('INSERT INTO vitals (heart_rate) VALUES (?), (?)').run(58, 61) + sources.push(database) +}) +const sources: Database.Database[] = [] +afterEach(() => { + for (const database of sources.splice(0)) database.close() + vi.unstubAllEnvs() + rmSync(root, { recursive: true, force: true }) +}) + +function archive(name: string, mtimeMs: number, body: string | Buffer = `frame ${name}`) { + const path = join(state.archive, name) + writeFileSync(path, body) + utimesSync(path, mtimeMs / 1000, mtimeMs / 1000) + return path +} + +async function loadRoute() { + vi.resetModules() + return await import('./route') +} + +const get = (GET: (request: Request) => Promise, query: string, init?: RequestInit) => + GET(new Request(`http://pod/api/biometrics/analysis-export?${query}`, init)) + +function untar(buffer: Buffer) { + const entries: Array<{ name: string, body: Buffer }> = [] + let position = 0 + for (;;) { + const header = parseTarHeader(buffer.subarray(position, position + 512)) + position += 512 + if (!header) break + entries.push({ name: header.name, body: 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') + +describe('GET /api/biometrics/analysis-export', () => { + it('selects archives by the widened mtime window and streams a hashed snapshot', async () => { + archive('A0000001.RAW.gz', START - 21 * MINUTE) + archive('A0000002.RAW.gz', START - 19 * MINUTE) + archive('A0000003.RAW.gz', START + 4 * 60 * MINUTE, Buffer.alloc(1300, 7)) + archive('A0000004.RAW.gz', END + 44 * MINUTE) + archive('A0000005.RAW.gz', END + 46 * MINUTE) + archive('A0000006.RAW.gz.tmp.4242', START + MINUTE) + archive('notes.txt', START + MINUTE) + symlinkSync(archive('linked-target', START), join(state.archive, 'A0000007.RAW.gz')) + const before = readdirSync(state.archive).sort() + + const { GET } = await loadRoute() + const response = await get(GET, `start=${new Date(START).toISOString()}&end=${END}`) + expect(response.status).toBe(200) + expect(response.headers.get('Content-Type')).toBe('application/x-tar') + const buffer = Buffer.from(await response.arrayBuffer()) + expect(buffer.length).toBe(Number(response.headers.get('Content-Length'))) + + const entries = untar(buffer) + expect(entries.map(entry => entry.name)).toEqual(['biometrics.db', 'raw/A0000002.RAW.gz', 'raw/A0000003.RAW.gz', 'raw/A0000004.RAW.gz', 'manifest.json']) + const manifest = JSON.parse(entries[4].body.toString()) + expect(manifest.formatVersion).toBe('analysis-export-v1') + expect(manifest.files).toEqual(entries.slice(0, 4).map(entry => expect.objectContaining({ path: entry.name, bytes: entry.body.length, sha256: sha256(entry.body) }))) + expect(manifest.files[2]).toEqual({ role: 'raw', path: 'raw/A0000003.RAW.gz', name: 'A0000003.RAW.gz', mtimeMs: START + 4 * 60 * MINUTE, bytes: 1300, sha256: sha256(Buffer.alloc(1300, 7)) }) + expect(manifest.window).toEqual({ + requested: { startMs: START, endMs: END, start: new Date(START).toISOString(), end: new Date(END).toISOString() }, + rawSelection: { includeRaw: true, mtimeFromMs: START - 20 * MINUTE, mtimeToMs: END + 45 * MINUTE }, + served: { archiveCount: 3, rawMtimeMinMs: START - 19 * MINUTE, rawMtimeMaxMs: END + 44 * MINUTE }, + }) + expect(manifest.pod).toEqual({ id: Buffer.from(hkdfSync('sha256', 'cid-0123', 'sleepypod-pod-id-v1', 'pod-id', 16)).toString('hex'), + idSource: 'mmc-cid', hostname: hostname(), model: { podVersion: 'J00', name: 'Pod 5', hardwareRevision: 'J55' } }) + expect(manifest.schema.biometrics).toEqual({ migrationHash: 'latest-hash', migrationCreatedAt: latest.when, migrationTag: latest.tag, appliedMigrations: 2 }) + expect(manifest.build).toEqual(expect.objectContaining({ branch: expect.any(String), commitHash: expect.any(String) })) + + const extracted = join(root, 'extracted.db') + writeFileSync(extracted, entries[0].body) + const copy = new Database(extracted, { readonly: true }) + expect(copy.prepare('SELECT heart_rate FROM vitals ORDER BY id').pluck().all()).toEqual([58, 61]) + copy.close() + + expect(readdirSync(snapshots)).toEqual([]) + expect(readdirSync(state.archive).sort()).toEqual(before) + }) + + it('omits RAW archives with raw=0', async () => { + archive('A0000001.RAW.gz', START) + const { GET } = await loadRoute() + const response = await get(GET, `start=${START}&end=${END}&raw=0`) + const entries = untar(Buffer.from(await response.arrayBuffer())) + expect(entries.map(entry => entry.name)).toEqual(['biometrics.db', 'manifest.json']) + const manifest = JSON.parse(entries[1].body.toString()) + expect(manifest.window.served).toEqual({ archiveCount: 0, rawMtimeMinMs: null, rawMtimeMaxMs: null }) + expect(manifest.window.rawSelection.includeRaw).toBe(false) + }) + + it('reports a null pod id when only a random dev seed exists', async () => { + state.seed = { source: 'random-dev', value: 'abc' } + try { + const { GET } = await loadRoute() + const entries = untar(Buffer.from(await (await get(GET, `start=${START}&end=${END}`)).arrayBuffer())) + expect(JSON.parse(String(entries.at(-1)?.body)).pod).toMatchObject({ id: null, idSource: 'random-dev' }) + } + finally { + state.seed = { source: 'mmc-cid', value: 'cid-0123' } + } + }) + + it.each([ + ['', 'missing start and end'], + [`start=${START}`, 'missing end'], + [`start=${END}&end=${START}`, 'start after end'], + [`start=${START}&end=${START}`, 'empty window'], + [`start=${START}&end=${START + 14 * 24 * 60 * MINUTE + 1}`, 'over 14 days'], + [`start=yesterday&end=${END}`, 'unparseable start'], + [`start=${START}&end=${END}&raw=yes`, 'bad raw flag'], + ])('rejects %s (%s) with 400', async (query) => { + const { GET } = await loadRoute() + const response = await get(GET, query) + expect(response.status).toBe(400) + expect((await response.json()).error).toMatch(/start and end/) + expect(readdirSync(snapshots)).toEqual([]) + }) + + it('accepts exactly 14 days', async () => { + const { GET } = await loadRoute() + const response = await get(GET, `start=${START}&end=${START + 14 * 24 * 60 * MINUTE}&raw=0`) + expect(response.status).toBe(200) + await response.arrayBuffer() + }) + + it('rejects windows over 2 GiB of RAW with 413 and no snapshot', async () => { + const sparse = archive('A0000001.RAW.gz', START, '') + truncateSync(sparse, 2 * 1024 ** 3 - 10) + utimesSync(sparse, START / 1000, START / 1000) + archive('A0000002.RAW.gz', START + MINUTE, Buffer.alloc(11)) + const { GET } = await loadRoute() + const response = await get(GET, `start=${START}&end=${END}`) + expect(response.status).toBe(413) + expect((await response.json()).error).toMatch(/2 RAW archives totalling 2147483649 bytes.*raw=0/) + expect(readdirSync(snapshots)).toEqual([]) + const without = await get(GET, `start=${START}&end=${END}&raw=0`) + expect(without.status).toBe(200) + await without.arrayBuffer() + }) + + it('allows one export at a time and removes the snapshot when the client cancels', async () => { + archive('A0000001.RAW.gz', START, Buffer.alloc(600_000, 1)) + const { GET } = await loadRoute() + const first = await get(GET, `start=${START}&end=${END}`) + expect(readdirSync(snapshots)).toHaveLength(1) + expect((await get(GET, `start=${START}&end=${END}`)).status).toBe(429) + const reader = (first.body as ReadableStream).getReader() + await reader.read() + await reader.cancel() + await vi.waitFor(() => expect(readdirSync(snapshots)).toEqual([])) + const next = await get(GET, `start=${START}&end=${END}&raw=0`) + expect(next.status).toBe(200) + await next.arrayBuffer() + }) + + it('removes the snapshot when the request is aborted mid-stream', async () => { + archive('A0000001.RAW.gz', START, Buffer.alloc(600_000, 1)) + const { GET } = await loadRoute() + const controller = new AbortController() + const response = await get(GET, `start=${START}&end=${END}`, { signal: controller.signal }) + await (response.body as ReadableStream).getReader().read() + controller.abort() + await vi.waitFor(() => expect(readdirSync(snapshots)).toEqual([])) + expect((await get(GET, `start=${START}&end=${END}&raw=0`)).status).toBe(200) + }) + + it('aborts the stream and removes the snapshot when an archive vanishes', async () => { + const path = archive('A0000001.RAW.gz', START) + const { GET } = await loadRoute() + const response = await get(GET, `start=${START}&end=${END}`) + unlinkSync(path) + await expect(response.arrayBuffer()).rejects.toThrow() + await vi.waitFor(() => expect(readdirSync(snapshots)).toEqual([])) + }) + + it('returns 500 and cleans up when the database cannot be read', async () => { + state.source = join(root, 'missing.db') + const error = vi.spyOn(console, 'error').mockImplementation(() => {}) + const { GET } = await loadRoute() + const response = await get(GET, `start=${START}&end=${END}`) + expect(response.status).toBe(500) + expect(error).toHaveBeenCalled() + expect(readdirSync(snapshots)).toEqual([]) + error.mockRestore() + }) +}) diff --git a/app/api/biometrics/analysis-export/route.ts b/app/api/biometrics/analysis-export/route.ts new file mode 100644 index 00000000..a5ab0b82 --- /dev/null +++ b/app/api/biometrics/analysis-export/route.ts @@ -0,0 +1,121 @@ +import { mkdtemp, rm, statfs } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { biometricsDb } from '@/src/db' +import { AnalysisExportTooLargeError, analysisWindowSchema, liveDatabaseBytes, planAnalysisExport } from '@/src/biometrics/analysis-export' +import { analysisBundleTarBytes, analysisExportChunks, podIdentity, readBuildInfo, snapshotBiometricsDatabase, type AnalysisBundle } from '@/src/biometrics/analysis-export-bundle' + +export const runtime = 'nodejs' +export const dynamic = 'force-dynamic' + +// One export at a time: each holds a database snapshot on tmpfs (RAM) until it finishes. +let inflight = false +const EXPORT_WATCHDOG_MS = (() => { + const raw = Number(process.env.ANALYSIS_EXPORT_WATCHDOG_MS ?? '') + return Number.isFinite(raw) && raw > 0 ? raw : 60 * 60 * 1000 +})() + +/** + * Stream an analysis snapshot as an uncompressed tar: a consistent backup of + * biometrics.db, the RAW archives for the window (raw=1, the default) and + * manifest.json with every file's SHA-256. Read-only: the backup goes through + * a read-only connection into os.tmpdir() (tmpfs on the pod) and is deleted + * when the stream ends, fails or the client disconnects. + */ +export async function GET(request: Request) { + const query = new URL(request.url).searchParams + const raw = query.get('raw') ?? '1' + const parsed = raw === '0' || raw === '1' + ? analysisWindowSchema.safeParse({ start: query.get('start') ?? '', end: query.get('end') ?? '', includeRaw: raw === '1' }) + : null + if (!parsed?.success) { + const detail = parsed ? parsed.error.issues.map(issue => issue.message).join('; ') : 'raw must be 0 or 1' + return Response.json({ error: `Specify start and end as ISO 8601 or epoch ms, at most 14 days apart, and raw=0|1: ${detail}` }, { status: 400 }) + } + if (inflight) return Response.json({ error: 'Another analysis export is in progress, retry later' }, { status: 429, headers: { 'Retry-After': '60' } }) + inflight = true + + let directory: string | undefined + const release = async () => { + const target = directory + directory = undefined + try { + if (target) await rm(target, { recursive: true, force: true }) + } + finally { inflight = false } + } + + let bundle: AnalysisBundle + try { + const plan = await planAnalysisExport(parsed.data) + const source = biometricsDb.$client.name + const neededBytes = await liveDatabaseBytes(source) + const space = await statfs(tmpdir()) + if (neededBytes > space.bavail * space.bsize) { + await release() + return Response.json({ error: `Not enough free space in ${tmpdir()} for a ${neededBytes}-byte database snapshot` }, { status: 507 }) + } + directory = await mkdtemp(path.join(tmpdir(), 'sp-analysis-export-')) + const [snapshot, build] = await Promise.all([snapshotBiometricsDatabase(source, directory), readBuildInfo()]) + bundle = { generatedAtMs: Date.now(), window: plan.window, database: { path: snapshot.path, bytes: snapshot.bytes }, + archives: plan.archives, build, pod: podIdentity(), schema: snapshot.schema } + } + catch (error) { + await release() + if (error instanceof AnalysisExportTooLargeError) return Response.json({ error: error.message }, { status: 413 }) + console.error('analysis export:', error) + return Response.json({ error: 'Analysis export failed' }, { status: 500 }) + } + + const chunks = analysisExportChunks(bundle) + let watchdog: ReturnType | undefined + let finished = false + // Runs once on completion, failure, cancel, client abort or the watchdog. + // chunks.return() closes any open file handle before the snapshot is removed. + const finish = async () => { + if (finished) return + finished = true + if (watchdog) clearTimeout(watchdog) + request.signal.removeEventListener('abort', onAbort) + try { + await chunks.return(undefined) + } + finally { await release() } + } + const onAbort = () => void finish() + request.signal.addEventListener('abort', onAbort, { once: true }) + const body = new ReadableStream({ + start(controller) { + watchdog = setTimeout(() => { + controller.error(new Error('Analysis export exceeded its time limit')) + void finish() + }, EXPORT_WATCHDOG_MS) + }, + async pull(controller) { + try { + const next = await chunks.next() + if (next.done) { + await finish() + controller.close() + } + else controller.enqueue(next.value) + } + catch (error) { + await finish() + controller.error(error) + } + }, + async cancel() { + await finish() + }, + }) + const { startMs, endMs } = bundle.window + return new Response(body, { + headers: { + 'Content-Type': 'application/x-tar', + 'Content-Disposition': `attachment; filename="sleepypod-analysis-${startMs}-${endMs}.tar"`, + 'Content-Length': String(analysisBundleTarBytes(bundle)), + 'Cache-Control': 'no-store', + }, + }) +} diff --git a/docs/biometrics-datasets.md b/docs/biometrics-datasets.md index 70fcfb31..1e0a456d 100644 --- a/docs/biometrics-datasets.md +++ b/docs/biometrics-datasets.md @@ -264,3 +264,59 @@ 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. + +## Analysis snapshot export + +`GET /api/biometrics/analysis-export?start=&end=&raw=1|0` +streams an uncompressed tar for off-pod analysis, such as comparing pod +biometrics with the Apple Watch. It replaces the ssh `backup()` + scp step and +`fetch_raw.sh`. The route is read-only and does no replay or backtest work: + +- `biometrics.db`: a consistent online backup taken through a separate + read-only connection. The backup is written to `os.tmpdir()` (tmpfs on the + pod, so the eMMC is never written) and deleted when the stream ends, fails or + the client disconnects. If the snapshot would not fit in tmpfs the route + returns 507. +- `raw/.RAW.gz` (when `raw=1`, the default): finished archives from + `/persistent/biometrics-archive` whose file mtime falls in + [start − 20 min, end + 45 min], oldest first. A frame is gzipped about 15 min + after it closes, so the widening catches the frames that overlap the window. + `*.tmp.*` partials and symlinks are skipped. +- `manifest.json`: the last entry, written once every other file has been + hashed during streaming. Fields: + - `formatVersion`: `analysis-export-v1`. + - `generatedAt` / `generatedAtMs` and `build` (branch, commit, title and build + date from `.git-info`). + - `pod`: `id` is a stable pod identifier, HKDF-SHA256 over the same + hardware-rooted seed chain the HomeKit identity uses (eMMC CID, then eMMC + serial, then `/etc/machine-id`; ADR 0020). It is the same on every export + and survives reinstalls; the seed itself never leaves the pod. `idSource` + names the seed used (`id` is null on a dev machine with no hardware seed). + `hostname`, and `model` (`podVersion`, model name and sensor hardware + revision from the last DAC status poll, or null before the first poll). + - `schema.biometrics`: the latest applied drizzle migration in the snapshot + (`migrationHash`, `migrationCreatedAt`, `migrationTag` from the migration + journal, `appliedMigrations`). + - `window`: `requested` start/end, `rawSelection` (the widened mtime range and + `includeRaw`) and `served` (archive count and the min/max mtime of the + archives actually included). + - `files`: every other entry with `bytes` and `sha256`; raw entries also carry + the original `name` and `mtimeMs`. + +The window must be at most 14 days with start before end (400 otherwise). If +the window's RAW archives total more than 2 GiB the route returns 413 without +taking a snapshot; narrow the window or pass `raw=0`. One export runs at a time +(429 while another is streaming). `Content-Length` is exact, so a download cut +short by a pruned archive or a dropped connection is detectable. + +The MCP tool `export_analysis_snapshot` (tRPC `biometricTuning.describeAnalysisExport`) +returns the download path, archive count, RAW bytes and an estimated tar size +for a window, never the bytes themselves. + +**Privacy:** the snapshot is the full biometrics database for both sides of the +bed plus raw sensor waveforms. Download it only to trusted machines and don't +share it. + +```sh +mkdir -p snapshot && curl -fsS "http://POD:3000/api/biometrics/analysis-export?start=2026-10-15T09:16:00Z&end=2026-10-15T15:24:00Z" | tar -xf - -C snapshot/ +``` diff --git a/src/biometrics/analysis-export-bundle.ts b/src/biometrics/analysis-export-bundle.ts new file mode 100644 index 00000000..4d6cb784 --- /dev/null +++ b/src/biometrics/analysis-export-bundle.ts @@ -0,0 +1,177 @@ +import { createHash, hkdfSync } from 'node:crypto' +import { constants, createReadStream } from 'node:fs' +import { open, readFile, stat } from 'node:fs/promises' +import { hostname } from 'node:os' +import { join, resolve } from 'node:path' +import Database from 'better-sqlite3' +import { getDacMonitorIfRunning } from '@/src/hardware/dacMonitor.instance' +import { POD_CAPS } from '@/src/hardware/pods' +import { readSeed } from '@/src/homekit/storage' +import { ANALYSIS_EXPORT_FORMAT, RAW_MTIME_LAG_MS, RAW_MTIME_LEAD_MS, tarBytes, type AnalysisWindow, type WindowArchive } from './analysis-export' +import { TAR_END, tarHeader, tarPadding } from './backtest/bundle-tar' + +export const ANALYSIS_DATABASE_PATH = 'biometrics.db' +export const ANALYSIS_MANIFEST_PATH = 'manifest.json' + +export interface SchemaVersion { migrationHash: string, migrationCreatedAt: number, migrationTag: string | null, appliedMigrations: number } +export interface PodIdentity { + id: string | null + idSource: ReturnType['source'] + hostname: string + model: { podVersion: string, name: string, hardwareRevision: string | null } | null +} +export interface BuildInfo { branch: string, commitHash: string, commitTitle: string, buildDate: string } + +export interface AnalysisBundle { + generatedAtMs: number + window: AnalysisWindow + database: { path: string, bytes: number } + archives: WindowArchive[] + build: BuildInfo + pod: PodIdentity + schema: SchemaVersion | null +} + +/** Latest applied drizzle migration of a biometrics database file. */ +export async function readSchemaVersion(databasePath: string, journalPath = resolve(process.cwd(), 'src/db/biometrics-migrations/meta/_journal.json')): Promise { + let row: { hash: string, createdAt: number, applied: number } | undefined + const database = new Database(databasePath, { readonly: true, fileMustExist: true }) + try { + row = database.prepare('SELECT hash, created_at AS createdAt, (SELECT COUNT(*) FROM __drizzle_migrations) AS applied FROM __drizzle_migrations ORDER BY created_at DESC, id DESC LIMIT 1') + .get() as typeof row + } + catch { + return null + } + finally { + database.close() + } + if (!row) return null + let migrationTag: string | null = null + try { + const journal = JSON.parse(await readFile(journalPath, 'utf-8')) as { entries?: Array<{ when?: unknown, tag?: unknown }> } + const entry = journal.entries?.find(item => item.when === Number(row.createdAt)) + if (typeof entry?.tag === 'string') migrationTag = entry.tag + } + catch { /* tag is provenance only */ } + return { migrationHash: row.hash, migrationCreatedAt: Number(row.createdAt), migrationTag, appliedMigrations: row.applied } +} + +/** Online backup through a separate read-only connection; never writes the source. */ +export async function snapshotBiometricsDatabase(sourcePath: string, directory: string) { + const path = join(directory, ANALYSIS_DATABASE_PATH) + const source = new Database(sourcePath, { readonly: true, fileMustExist: true }) + try { + await source.backup(path) + } + finally { + source.close() + } + return { path, bytes: (await stat(path)).size, schema: await readSchemaVersion(path) } +} + +/** + * Same hardware-rooted seed chain as the HomeKit identity (ADR 0020), so the id + * survives reinstalls and /persistent wipes. Derived through HKDF so the eMMC + * CID or serial itself never leaves the pod. Null when only a random dev seed exists. + */ +export function podIdentity(): PodIdentity { + const seed = readSeed() + const id = seed.source === 'random-dev' ? null : Buffer.from(hkdfSync('sha256', seed.value, 'sleepypod-pod-id-v1', 'pod-id', 16)).toString('hex') + const status = getDacMonitorIfRunning()?.getLastStatus() + const caps = status ? POD_CAPS[status.podVersion] : undefined + const model = status && caps + ? { podVersion: status.podVersion, name: caps.modelName, + hardwareRevision: status.sensorLabel.split('-').reverse().find(segment => /^[A-Z]\d{2}$/.test(segment)) ?? null } + : null + return { id, idSource: seed.source, hostname: hostname(), model } +} + +export async function readBuildInfo(): Promise { + const info = { branch: 'unknown', commitHash: 'unknown', commitTitle: 'unknown', buildDate: 'unknown' } + try { + const parsed = JSON.parse(await readFile('.git-info', 'utf-8')) as Record + for (const key of Object.keys(info) as Array) { + if (typeof parsed[key] === 'string') info[key] = parsed[key] + } + } + catch { /* not built from a checkout */ } + return info +} + +const PLACEHOLDER_HASH = '0'.repeat(64) + +/** manifest.json: provenance plus every other file with its size and SHA-256. */ +export function analysisManifest(bundle: AnalysisBundle, hashes: { database: string, raw: readonly string[] }) { + const { window, archives } = bundle + const mtimes = archives.map(archive => archive.mtimeMs) + return { + formatVersion: ANALYSIS_EXPORT_FORMAT, + generatedAt: new Date(bundle.generatedAtMs).toISOString(), + generatedAtMs: bundle.generatedAtMs, + build: bundle.build, + pod: bundle.pod, + schema: { biometrics: bundle.schema }, + window: { + requested: { startMs: window.startMs, endMs: window.endMs, start: new Date(window.startMs).toISOString(), end: new Date(window.endMs).toISOString() }, + rawSelection: { includeRaw: window.includeRaw, mtimeFromMs: window.startMs - RAW_MTIME_LEAD_MS, mtimeToMs: window.endMs + RAW_MTIME_LAG_MS }, + served: { archiveCount: archives.length, rawMtimeMinMs: mtimes.length ? Math.min(...mtimes) : null, rawMtimeMaxMs: mtimes.length ? Math.max(...mtimes) : null }, + }, + files: [ + { role: 'database', path: ANALYSIS_DATABASE_PATH, bytes: bundle.database.bytes, sha256: hashes.database }, + ...archives.map((archive, index) => ({ role: 'raw', path: `raw/${archive.name}`, name: archive.name, mtimeMs: archive.mtimeMs, + bytes: archive.bytes, sha256: hashes.raw[index] })), + ], + privacy: 'Contains the full biometrics database for both sides of the bed. Keep it on trusted machines.', + } +} + +// Hashes are fixed-width hex, so a placeholder manifest has the final length and +// Content-Length can be sent before the files are read. +function manifestBytes(bundle: AnalysisBundle) { + return Buffer.byteLength(JSON.stringify(analysisManifest(bundle, { database: PLACEHOLDER_HASH, raw: bundle.archives.map(() => PLACEHOLDER_HASH) }))) +} + +/** Exact byte length of the tar analysisExportChunks yields. */ +export function analysisBundleTarBytes(bundle: AnalysisBundle) { + return tarBytes([bundle.database.bytes, ...bundle.archives.map(archive => archive.bytes), manifestBytes(bundle)]) +} + +/** Stream one file as a tar entry, hashing on the way; it must keep its listed size. */ +async function* fileEntry(path: string, name: string, bytes: number, mtimeSeconds: number, digests: string[]) { + const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW) + try { + const info = await handle.stat() + if (!info.isFile() || info.size !== bytes) throw new Error(`${name} changed size before export`) + yield tarHeader(name, bytes, mtimeSeconds) + const digest = createHash('sha256') + let sent = 0 + if (bytes > 0) { + for await (const chunk of createReadStream('', { fd: handle.fd, autoClose: false, start: 0, end: bytes - 1, highWaterMark: 256 * 1024 })) { + digest.update(chunk as Buffer) + sent += (chunk as Buffer).length + yield chunk as Buffer + } + } + if (sent !== bytes) throw new Error(`${name} shrank during export`) + digests.push(digest.digest('hex')) + yield tarPadding(bytes) + } + finally { + await handle.close() + } +} + +/** biometrics.db, raw/.RAW.gz in mtime order, then manifest.json with the hashes taken while streaming. */ +export async function* analysisExportChunks(bundle: AnalysisBundle) { + const generatedSeconds = bundle.generatedAtMs / 1000 + const digests: string[] = [] + yield* fileEntry(bundle.database.path, ANALYSIS_DATABASE_PATH, bundle.database.bytes, generatedSeconds, digests) + for (const archive of bundle.archives) yield* fileEntry(archive.path, `raw/${archive.name}`, archive.bytes, archive.mtimeMs / 1000, digests) + const manifest = Buffer.from(JSON.stringify(analysisManifest(bundle, { database: digests[0], raw: digests.slice(1) }))) + if (manifest.length !== manifestBytes(bundle)) throw new Error('Manifest length changed while streaming') + yield tarHeader(ANALYSIS_MANIFEST_PATH, manifest.length, generatedSeconds) + yield manifest + yield tarPadding(manifest.length) + yield TAR_END +} diff --git a/src/biometrics/analysis-export.ts b/src/biometrics/analysis-export.ts new file mode 100644 index 00000000..00e3b330 --- /dev/null +++ b/src/biometrics/analysis-export.ts @@ -0,0 +1,93 @@ +import { lstat, readdir, stat } from 'node:fs/promises' +import { join } from 'node:path' +import { z } from 'zod' +import { ARCHIVE_DIR } from '@/src/lib/podStorage' + +export const ANALYSIS_EXPORT_FORMAT = 'analysis-export-v1' +export const ANALYSIS_EXPORT_PATH = '/api/biometrics/analysis-export' +export const MAX_ANALYSIS_WINDOW_MS = 14 * 24 * 60 * 60 * 1000 +export const MAX_ANALYSIS_RAW_BYTES = 2 * 1024 ** 3 +// A frame's archive is written ~15 min after the frame closes, so a frame that +// starts just before the window is stamped up to 20 min later, and the last +// frame of the window up to 45 min after its end. +export const RAW_MTIME_LEAD_MS = 20 * 60 * 1000 +export const RAW_MTIME_LAG_MS = 45 * 60 * 1000 + +// Only finished archives: the archiver writes .RAW.gz.tmp. and renames. +const ARCHIVE_NAME = /^[A-Za-z0-9_-]+\.RAW\.gz$/ + +const instant = z.string().trim().min(1).max(40) +export const analysisExportRequestSchema = z.object({ start: instant, end: instant, includeRaw: z.boolean().default(true) }).strict() +export type AnalysisExportRequest = z.input + +/** Start/end as epoch ms or ISO 8601, start < end, at most 14 days apart. */ +export const analysisWindowSchema = analysisExportRequestSchema.transform((input, ctx) => { + const toMs = (value: string, key: 'start' | 'end') => { + const ms = /^\d+$/.test(value) ? Number(value) : Date.parse(value) + if (Number.isSafeInteger(ms) && ms >= 0) return ms + ctx.addIssue({ code: 'custom', path: [key], message: `${key} must be an ISO 8601 timestamp or epoch milliseconds` }) + return null + } + const startMs = toMs(input.start, 'start'), endMs = toMs(input.end, 'end') + if (startMs === null || endMs === null) return z.NEVER + if (startMs >= endMs) ctx.addIssue({ code: 'custom', path: ['end'], message: 'start must be before end' }) + else if (endMs - startMs > MAX_ANALYSIS_WINDOW_MS) ctx.addIssue({ code: 'custom', path: ['end'], message: 'The window must be 14 days or shorter' }) + return { startMs, endMs, includeRaw: input.includeRaw } +}) +export type AnalysisWindow = z.output + +export class AnalysisExportTooLargeError extends Error { + constructor(readonly rawBytes: number, readonly archiveCount: number) { + super(`The window holds ${archiveCount} RAW archives totalling ${rawBytes} bytes, over the ${MAX_ANALYSIS_RAW_BYTES}-byte limit. Narrow the window or pass raw=0.`) + } +} + +export interface WindowArchive { name: string, path: string, bytes: number, mtimeMs: number } + +/** Finished RAW archives whose mtime falls in [start − 20 min, end + 45 min], oldest first. */ +export async function listWindowArchives(window: Pick, directory = ARCHIVE_DIR): Promise { + let names: string[] + try { + names = await readdir(directory) + } + catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return [] + throw error + } + const from = window.startMs - RAW_MTIME_LEAD_MS, to = window.endMs + RAW_MTIME_LAG_MS + const archives: WindowArchive[] = [] + for (const name of names) { + if (!ARCHIVE_NAME.test(name)) continue + const path = join(directory, name) + // The pruner may delete an archive between readdir and lstat. + const info = await lstat(path).catch(() => null) + if (!info?.isFile()) continue + const mtimeMs = Math.floor(info.mtimeMs) + if (mtimeMs < from || mtimeMs > to) continue + archives.push({ name, path, bytes: info.size, mtimeMs }) + } + return archives.sort((a, b) => a.mtimeMs - b.mtimeMs || a.name.localeCompare(b.name)) +} + +/** Archives to send for a window; throws AnalysisExportTooLargeError over the cap. */ +export async function planAnalysisExport(window: AnalysisWindow, directory = ARCHIVE_DIR) { + const archives = window.includeRaw ? await listWindowArchives(window, directory) : [] + const rawBytes = archives.reduce((sum, archive) => sum + archive.bytes, 0) + if (rawBytes > MAX_ANALYSIS_RAW_BYTES) throw new AnalysisExportTooLargeError(rawBytes, archives.length) + return { window, archives, rawBytes } +} + +/** Live database plus its WAL: an upper bound for the size of its backup. */ +export async function liveDatabaseBytes(file: string) { + const sizes = await Promise.all([file, `${file}-wal`].map(name => stat(name).then(info => info.size, () => 0))) + return sizes[0] + sizes[1] +} + +/** Uncompressed ustar size: a header per entry, 512-byte padding, two end blocks. */ +export function tarBytes(entryBytes: readonly number[]) { + return entryBytes.reduce((sum, bytes) => sum + 512 + Math.ceil(bytes / 512) * 512, 1024) +} + +export function analysisExportDownloadPath(window: AnalysisWindow) { + return `${ANALYSIS_EXPORT_PATH}?start=${window.startMs}&end=${window.endMs}&raw=${window.includeRaw ? 1 : 0}` +} diff --git a/src/biometrics/tests/analysis-export.test.ts b/src/biometrics/tests/analysis-export.test.ts new file mode 100644 index 00000000..e1cb9af9 --- /dev/null +++ b/src/biometrics/tests/analysis-export.test.ts @@ -0,0 +1,70 @@ +// @vitest-environment node +import { mkdtempSync, rmSync, truncateSync, utimesSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { AnalysisExportTooLargeError, analysisExportDownloadPath, analysisWindowSchema, listWindowArchives, planAnalysisExport, tarBytes } from '../analysis-export' + +const START = Date.UTC(2026, 9, 10, 6) +const END = Date.UTC(2026, 9, 10, 14) +let directory: string +beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), 'analysis-window-')) +}) +afterEach(() => rmSync(directory, { recursive: true, force: true })) + +describe('analysisWindowSchema', () => { + it('accepts ISO 8601 and epoch ms and defaults includeRaw', () => { + expect(analysisWindowSchema.parse({ start: '2026-10-10T06:00:00Z', end: String(END) })).toEqual({ startMs: START, endMs: END, includeRaw: true }) + expect(analysisWindowSchema.parse({ start: String(START), end: '2026-10-10T07:00:00-07:00', includeRaw: false })).toEqual({ startMs: START, endMs: END, includeRaw: false }) + }) + + it.each([ + [{ start: 'soon', end: String(END) }, /start must be an ISO 8601/], + [{ start: String(END), end: String(START) }, /start must be before end/], + [{ start: String(START), end: String(START + 14 * 86_400_000 + 1) }, /14 days or shorter/], + [{ start: String(START), end: String(END), extra: 1 }, /Unrecognized key/], + ])('rejects %j', (input, message) => { + const result = analysisWindowSchema.safeParse(input) + expect(result.success).toBe(false) + expect(result.error?.issues.map(issue => issue.message).join('; ')).toMatch(message) + }) +}) + +describe('planAnalysisExport', () => { + const at = (name: string, mtimeMs: number, bytes = 10) => { + const path = join(directory, name) + writeFileSync(path, Buffer.alloc(bytes)) + utimesSync(path, mtimeMs / 1000, mtimeMs / 1000) + } + + it('includes the inclusive [start − 20 min, end + 45 min] mtime range, oldest first', async () => { + at('B.RAW.gz', END + 45 * 60_000) + at('A.RAW.gz', START - 20 * 60_000) + at('C.RAW.gz', START - 20 * 60_000 - 1000) + at('D.RAW.gz', END + 45 * 60_000 + 1000) + expect((await listWindowArchives({ startMs: START, endMs: END }, directory)).map(archive => archive.name)).toEqual(['A.RAW.gz', 'B.RAW.gz']) + }) + + it('treats a missing archive directory as empty and skips listing when RAW is off', async () => { + expect(await listWindowArchives({ startMs: START, endMs: END }, join(directory, 'missing'))).toEqual([]) + at('A.RAW.gz', START) + expect(await planAnalysisExport({ startMs: START, endMs: END, includeRaw: false }, directory)).toMatchObject({ archives: [], rawBytes: 0 }) + }) + + it('sums RAW bytes and rejects past the cap', async () => { + at('A.RAW.gz', START, 100) + at('B.RAW.gz', START + 1000, 28) + expect(await planAnalysisExport({ startMs: START, endMs: END, includeRaw: true }, directory)).toMatchObject({ rawBytes: 128 }) + const sparse = join(directory, 'C.RAW.gz') + writeFileSync(sparse, '') + truncateSync(sparse, 2 * 1024 ** 3) + utimesSync(sparse, START / 1000, START / 1000) + await expect(planAnalysisExport({ startMs: START, endMs: END, includeRaw: true }, directory)).rejects.toBeInstanceOf(AnalysisExportTooLargeError) + }) +}) + +it('sizes the tar and builds the download path', () => { + expect(tarBytes([0, 1, 512, 513])).toBe(1024 + 4 * 512 + 0 + 512 + 512 + 1024) + expect(analysisExportDownloadPath({ startMs: START, endMs: END, includeRaw: false })).toBe(`/api/biometrics/analysis-export?start=${START}&end=${END}&raw=0`) +}) diff --git a/src/demo/handlers/biometricTuning.ts b/src/demo/handlers/biometricTuning.ts index e336a9a3..34538c64 100644 --- a/src/demo/handlers/biometricTuning.ts +++ b/src/demo/handlers/biometricTuning.ts @@ -159,6 +159,7 @@ export const biometricTuning: DemoHandlers<'biometricTuning'> = { { label: 'candidate', revisionId, manifestHash: candidate.manifestHash, resolvedHash: candidate.hash, resolved: candidate.resolved }, ] }), + describeAnalysisExport: () => { throw new Error('SIMULATED DEMO: analysis snapshots need a real pod database and RAW archives.') }, getManifest: async () => ({ manifest: biometricManifest, manifestHash: await hash(biometricManifest), activationAvailable: true }), resolveProfile: input => resolve(input.overrides, input.baseManifestHash), createProfileRevision: createDemoRevision, diff --git a/src/lib/podStorage.ts b/src/lib/podStorage.ts index 3a7b4c5e..6b73879a 100644 --- a/src/lib/podStorage.ts +++ b/src/lib/podStorage.ts @@ -15,7 +15,7 @@ const execFileAsync = promisify(execFile) export const PERSISTENT_ROOT = '/persistent' export const PRUNER_TARGET_PERCENT = Number(process.env.SP_PRUNER_TARGET_PCT) || 80 -const ARCHIVE_DIR = `${PERSISTENT_ROOT}/biometrics-archive` +export const ARCHIVE_DIR = `${PERSISTENT_ROOT}/biometrics-archive` const SWAPFILE = `${PERSISTENT_ROOT}/swapfile` const INSTALL_DIR = process.env.SP_INSTALL_DIR ?? '/home/dac/sleepypod-core' const CLEANUP_BIN = '/usr/local/bin/sp-storage-cleanup' diff --git a/src/mcp/README.md b/src/mcp/README.md index 909e4b81..79e92d34 100644 --- a/src/mcp/README.md +++ b/src/mcp/README.md @@ -142,6 +142,7 @@ app/api/mcp/route.ts Next.js route handler (stateless Streamable HTTP) | `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) | +| `export_analysis_snapshot` | Count the RAW archives for a window (≤ 14 days) and return the LAN download path for a tar with a `biometrics.db` backup, those archives and a hashed `manifest.json` (no 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 3c7828ad..8f068503 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(), describeDatasetExport: fn(), listDatasets: fn(), createExperiment: fn(), + createProfileRevision: fn(), createDataset: fn(), getDataset: fn(), describeDatasetExport: fn(), describeAnalysisExport: 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(), }, @@ -118,6 +118,7 @@ describe('sleepypod MCP server', () => { 'apply_sleep_curve', 'backtest_biometrics', 'diagnose_pod', + 'export_analysis_snapshot', 'export_calibration_dataset', 'get_automations', 'get_bed_occupancy', @@ -1246,6 +1247,15 @@ it('routes every remaining inspection, experiment, backtest and release action t expect(api.describeDatasetExport).toHaveBeenCalledWith({ id }) expect((await client.listTools()).tools.find(tool => tool.name === 'export_calibration_dataset')?.annotations).toMatchObject({ readOnlyHint: true, destructiveHint: false }) + const snapshot = { window: { startMs: 1791612000000, endMs: 1791640800000, includeRaw: true }, archiveCount: 3, rawBytes: 900, maxRawBytes: 2 * 1024 ** 3, estimatedBytes: 4096, + download: { method: 'GET', path: '/api/biometrics/analysis-export?start=1791612000000&end=1791640800000&raw=1', contentType: 'application/x-tar' } } + api.describeAnalysisExport.mockResolvedValue(snapshot) + expect(await call('export_analysis_snapshot', { start: '2026-10-10T06:00:00.000Z', end: '1791640800000' })).toEqual(snapshot) + expect(api.describeAnalysisExport).toHaveBeenCalledWith({ start: '2026-10-10T06:00:00.000Z', end: '1791640800000', includeRaw: true }) + await call('export_analysis_snapshot', { start: '1791612000000', end: '1791640800000', includeRaw: false }) + expect(api.describeAnalysisExport).toHaveBeenLastCalledWith({ start: '1791612000000', end: '1791640800000', includeRaw: false }) + expect((await client.listTools()).tools.find(tool => tool.name === 'export_analysis_snapshot')?.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 0f15bc12..64c4989e 100644 --- a/src/mcp/tools/biometrics.ts +++ b/src/mcp/tools/biometrics.ts @@ -16,6 +16,7 @@ import { traceRequestSchema } from '@/src/biometrics/backtest/trace-contracts' import { submitBacktestSchema } from '@/src/biometrics/backtest/submission' import { submitShadowSchema } from '@/src/biometrics/shadow/submission-contracts' import { prepareShadowRequestSchema } from '@/src/biometrics/shadow/preparation-contracts' +import { analysisExportRequestSchema } from '@/src/biometrics/analysis-export' const side = z.enum(['left', 'right']) const id = z.object({ id: z.string().uuid() }).strict() @@ -151,6 +152,12 @@ export function registerBiometricTools(server: McpServer) { }, 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('export_analysis_snapshot', { + title: 'Locate an analysis snapshot of the biometrics database and RAW archives', + description: 'Count the RAW archives for a window (start/end as ISO 8601 or epoch ms, at most 14 days) and return the LAN download path for a tar holding a consistent biometrics.db backup, those archives and a hashed manifest.json, with the archive count and estimated bytes. Bytes are never placed in this response. Windows over 2 GiB of RAW are rejected; narrow the window or set includeRaw false. The snapshot holds the full biometrics database for both sides: download only to a user-authorized machine.', + inputSchema: analysisExportRequestSchema.shape, annotations: read, + }, input => runTool(async () => jsonResult(await getCaller().biometricTuning.describeAnalysisExport(input)))) + 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 4d528267..1a06b4f5 100644 --- a/src/server/routers/biometricTuning.ts +++ b/src/server/routers/biometricTuning.ts @@ -33,6 +33,7 @@ import { describeJob, jobDescriptorSchema, resultDescriptorSchema } from '@/src/ import { experimentProtocolSchema } from '@/src/biometrics/backtest/contracts' import { datasetDescriptorSchema } from '@/src/biometrics/backtest/contracts' import { datasetBundleIndexSchema, datasetBundleTarBytes } from '@/src/biometrics/backtest/dataset-export' +import { AnalysisExportTooLargeError, MAX_ANALYSIS_RAW_BYTES, analysisExportDownloadPath, analysisExportRequestSchema, analysisWindowSchema, liveDatabaseBytes, planAnalysisExport, tarBytes } from '@/src/biometrics/analysis-export' import { deviceSettings } from '@/src/db/schema' import { ReferenceInputError } from '@/src/biometrics/references/time' import { countUnlinkedLegacyReferences, linkLegacyReferences } from '@/src/biometrics/references/legacy' @@ -295,6 +296,29 @@ export const biometricTuningRouter = router({ .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) } })))), + describeAnalysisExport: publicProcedure + .meta({ openapi: { method: 'GET', path: '/biometric-tuning/analysis-export-index', protect: false, tags: ['Biometric tuning'] } }) + .input(analysisExportRequestSchema) + .output(z.object({ window: z.object({ startMs: z.number(), endMs: z.number(), includeRaw: z.boolean() }).strict(), archiveCount: z.number().int().nonnegative(), + rawBytes: z.number().int().nonnegative(), maxRawBytes: z.number().int().positive(), estimatedBytes: z.number().int().nonnegative(), + download: z.object({ method: z.literal('GET'), path: z.string(), contentType: z.literal('application/x-tar') }).strict() }).strict()) + .query(async ({ input }) => { + const parsed = analysisWindowSchema.safeParse(input) + if (!parsed.success) throw new TRPCError({ code: 'BAD_REQUEST', message: parsed.error.issues.map(issue => issue.message).join('; ') }) + try { + const plan = await planAnalysisExport(parsed.data) + const databaseBytes = await liveDatabaseBytes(biometricsDb.$client.name) + return { window: plan.window, archiveCount: plan.archives.length, rawBytes: plan.rawBytes, maxRawBytes: MAX_ANALYSIS_RAW_BYTES, + // The manifest is a few hundred bytes per archive. + estimatedBytes: tarBytes([databaseBytes, ...plan.archives.map(archive => archive.bytes), 2048 + 300 * plan.archives.length]), + download: { method: 'GET' as const, path: analysisExportDownloadPath(plan.window), contentType: 'application/x-tar' as const } } + } + catch (error) { + if (error instanceof AnalysisExportTooLargeError) throw new TRPCError({ code: 'PAYLOAD_TOO_LARGE', message: error.message, cause: error }) + throw error + } + }), + listDatasets: publicProcedure .meta({ openapi: { method: 'GET', path: '/biometric-tuning/datasets', protect: false, tags: ['Biometric tuning'] } }) .input(z.object({ side: sideSchema, beforeId: z.string().uuid().optional(), beforeCreatedAtMs: z.number().int().nonnegative().optional() }).strict() From 03079b1d085a150c66ef91fd2dc9231783caaee3 Mon Sep 17 00:00:00 2001 From: Jonathan Ng Date: Sun, 18 Oct 2026 11:23:28 -0700 Subject: [PATCH 2/2] fix(biometrics): clean up an analysis export whose client left during the snapshot, and forbid transforming the tar --- app/api/biometrics/analysis-export/route.test.ts | 12 ++++++++++++ app/api/biometrics/analysis-export/route.ts | 9 ++++++++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/app/api/biometrics/analysis-export/route.test.ts b/app/api/biometrics/analysis-export/route.test.ts index 84170f5a..b7a25d63 100644 --- a/app/api/biometrics/analysis-export/route.test.ts +++ b/app/api/biometrics/analysis-export/route.test.ts @@ -98,6 +98,7 @@ describe('GET /api/biometrics/analysis-export', () => { const response = await get(GET, `start=${new Date(START).toISOString()}&end=${END}`) expect(response.status).toBe(200) expect(response.headers.get('Content-Type')).toBe('application/x-tar') + expect(response.headers.get('Cache-Control')).toBe('no-store, no-transform') const buffer = Buffer.from(await response.arrayBuffer()) expect(buffer.length).toBe(Number(response.headers.get('Content-Length'))) @@ -214,6 +215,17 @@ describe('GET /api/biometrics/analysis-export', () => { expect((await get(GET, `start=${START}&end=${END}&raw=0`)).status).toBe(200) }) + it('cleans up and streams nothing when the client left while the snapshot was taken', async () => { + archive('A0000001.RAW.gz', START) + const { GET } = await loadRoute() + const controller = new AbortController() + controller.abort() + const response = await get(GET, `start=${START}&end=${END}`, { signal: controller.signal }) + expect(response.status).toBe(499) + expect(readdirSync(snapshots)).toEqual([]) + expect((await get(GET, `start=${START}&end=${END}&raw=0`)).status).toBe(200) + }) + it('aborts the stream and removes the snapshot when an archive vanishes', async () => { const path = archive('A0000001.RAW.gz', START) const { GET } = await loadRoute() diff --git a/app/api/biometrics/analysis-export/route.ts b/app/api/biometrics/analysis-export/route.ts index a5ab0b82..48a0ef83 100644 --- a/app/api/biometrics/analysis-export/route.ts +++ b/app/api/biometrics/analysis-export/route.ts @@ -84,6 +84,12 @@ export async function GET(request: Request) { } const onAbort = () => void finish() request.signal.addEventListener('abort', onAbort, { once: true }) + // The client may have gone while the snapshot was being taken; the abort + // event has then already fired, so clean up now instead of streaming. + if (request.signal.aborted) { + await finish() + return new Response(null, { status: 499 }) + } const body = new ReadableStream({ start(controller) { watchdog = setTimeout(() => { @@ -115,7 +121,8 @@ export async function GET(request: Request) { 'Content-Type': 'application/x-tar', 'Content-Disposition': `attachment; filename="sleepypod-analysis-${startMs}-${endMs}.tar"`, 'Content-Length': String(analysisBundleTarBytes(bundle)), - 'Cache-Control': 'no-store', + // no-transform: a proxy that recompresses the tar would break Content-Length and the manifest hashes. + 'Cache-Control': 'no-store, no-transform', }, }) }