CR: fix/personal-model-safety-1022

.github/workflows/python-modules.yml

@@ -83,8 +83,8 @@ jobs:
             target: .
             cov_source: .
             deps: ""
-          # Stdlib-only repo scripts; tests load them by path from
-          # scripts/tests/.
+          # Repo scripts, including the NumPy-backed personal-model label helper.
+          # Reuse the locked numerical environment for reproducible tests.
           - name: scripts
             workdir: scripts
             target: tests
@@ -113,6 +113,8 @@ jobs:
           if [ "${{ matrix.module.name }}" = "common" ]; then
             # Keep common's cwd/import layout while using the numerical lock.
             UV_PROJECT_ENVIRONMENT="$PWD/.venv" uv sync --project piezo-processor --locked --no-install-project
+          elif [ "${{ matrix.module.name }}" = "scripts" ]; then
+            UV_PROJECT_ENVIRONMENT="$PWD/.venv" uv sync --project ../modules/piezo-processor --locked --no-install-project
           elif [ -f uv.lock ]; then
             uv sync --locked --no-install-project
           else

docs/research/sleep-stage-calibration/README.md

@@ -205,20 +205,24 @@ Two parity checks keep the container and the phone honest:
 `scripts/train-personal-models.py` chains the whole thing: snapshot the pod's
 database over SSH, build the dataset, train both models on a contract with
 leave-one-night-out scoring, export the pod form, and upload the model with
-the higher pooled κ. About five seconds on Apple silicon; nothing runs on the
+the higher pooled κ as an unselected candidate, then print its preview. About five seconds on Apple silicon; nothing runs on the
 phone.
 
 ```sh
-scripts/train-personal-models.py --pod <pod-ip> --side left            # contract B, best model, uploaded
+scripts/train-personal-models.py --pod <pod-ip> --side left            # contract B, candidate stored and previewed
 scripts/train-personal-models.py --pod <pod-ip> --side left --no-upload --contract A
 scripts/train-personal-models.py --pod <pod-ip> --snapshot poddb/biometrics.db   # reuse a snapshot
 ```
 
 The pod needs the model endpoints below, and the Watch stages must already be
 on it (section 2b). The run leaves the dataset and the exported models (about
 half a megabyte) in the work directory and deletes the ~75 MB database
-snapshot unless `--keep-snapshot`; the pod keeps one model row per side,
-replaced on every upload.
+snapshot unless `--keep-snapshot`. Candidates accumulate in the model library; the
+current selection and activation stay unchanged. `--select` explicitly selects the
+candidate inactive (turning personal staging off); `--activate` selects it active,
+after preview succeeds. The two flags are mutually exclusive; neither can be
+combined with `--no-upload`. Preview
+scores on training nights are in-sample, not held-out validation.
 
 ### Running a model on the pod
 
@@ -232,13 +236,19 @@ same synthetic fixture is replayed by `src/lib/tests/personal-stage-model.test.t
 and refuses a model that fails its own test vectors.
 
 ```sh
-# store the container's forest on the pod for the left side
-curl -X PUT http://<pod-ip>:3000/api/sleep-stages/model   -H 'content-type: application/json'   -d "$(jq -c '{side:"left", model:.}' out/personal-forest.json)"
+# store a candidate without changing the selection
+curl -X POST http://<pod-ip>:3000/api/sleep-stages/models   -H 'content-type: application/json'   -d "$(jq -c '{side:"left", model:.}' out/personal-forest.json)"
 # score it beside the rules against the Watch on the last 14 nights
-curl -X POST http://<pod-ip>:3000/api/sleep-stages/model/preview   -H 'content-type: application/json' -d '{"side":"left","nights":14}'
+curl -X POST http://<pod-ip>:3000/api/sleep-stages/model/preview   -H 'content-type: application/json' -d "$(jq -c '{side:"left", nights:14, model:.}' out/personal-forest.json)"
 ```
 
-The stored model can also stage the side's nights: `PUT /api/sleep-stages/model/active`
+The legacy singular `PUT /api/sleep-stages/model` replaces the selected model in
+place and preserves its active state, so it immediately changes staging on an
+active side. It archives both the previous selection and the uploaded model in
+the library. To select a library candidate explicitly, use
+`PUT /api/sleep-stages/models/selected` with `{side, id, active}`.
+
+The selected model can also stage the side's nights: `PUT /api/sleep-stages/model/active`
 with `{"side":"left","active":true}` (or the "Stage nights with this model"
 switch on the Settings → Sleep stages page) makes `getSleepStages` and the
 Watch comparison classify with the model, with the profile's rules as the

docs/research/sleep-stage-calibration/personal-models.py

@@ -123,17 +123,20 @@ def window(samples, mid, half):
 def label(epochs, rows):
     """The Watch stage of the labeled row containing the epoch midpoint (first match), as StageDataset.label."""
     segments = []
+    interrupted = False
     for r in rows:
         t = int(r['ts'])
         if not r['watch_stage']:
             # An explicit unlabeled row ends the running segment; a missing sample (gap) is bridged below.
             if segments:
                 segments[-1][1] = min(segments[-1][1], t)
+            interrupted = True
             continue
-        if segments and segments[-1][2] == r['watch_stage'] and 0 <= t - segments[-1][1] <= 120:
+        if not interrupted and segments and segments[-1][2] == r['watch_stage'] and 0 <= t - segments[-1][1] <= 120:
             segments[-1][1] = t + 60
         else:
             segments.append([t, t + 60, r['watch_stage']])
+        interrupted = False
     out = []
     for e in epochs:
         mid = e['start'] + EPOCH / 2

scripts/tests/test_personal_models.py

@@ -0,0 +1,109 @@
+"""Synthetic regressions for candidate uploads and explicit Watch label gaps."""
+import importlib.util
+import io
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def load(name, path):
+    spec = importlib.util.spec_from_file_location(name, ROOT / path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+train = load('train_personal', 'scripts/train-personal-models.py')
+models = load('personal', 'docs/research/sleep-stage-calibration/personal-models.py')
+
+
+class CandidateTests(unittest.TestCase):
+    def test_upload_routes_and_explicit_selection(self):
+        for select, activate in [(False, False), (True, False), (False, True)]:
+            with self.subTest(select=select, activate=activate), tempfile.TemporaryDirectory() as tmp:
+                path = Path(tmp) / 'model.json'
+                path.write_text('{"version":"test"}')
+                calls = []
+
+                def response(request, timeout):
+                    calls.append((request.method, request.full_url, json.loads(request.data)))
+                    return io.StringIO('{"id":7}')
+
+                with patch.object(train.urllib.request, 'urlopen', side_effect=response), patch('sys.stdout', new_callable=io.StringIO) as output:
+                    train.upload('pod', 'left', path, select=select, activate=activate)
+                self.assertEqual(calls[0], ('POST', 'http://pod:3000/api/sleep-stages/models', {'side': 'left', 'model': {'version': 'test'}}))
+                self.assertEqual(calls[1], ('POST', 'http://pod:3000/api/sleep-stages/model/preview', {'side': 'left', 'nights': 14, 'model': {'version': 'test'}}))
+                self.assertIn('in-sample', output.getvalue())
+                self.assertEqual(len(calls), 3 if select or activate else 2)
+                if select or activate:
+                    self.assertEqual(calls[2], ('PUT', 'http://pod:3000/api/sleep-stages/models/selected', {'side': 'left', 'id': 7, 'active': activate}))
+
+    def test_failed_preview_never_selects(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / 'model.json'
+            path.write_text('{}')
+            with patch.object(train, 'request_json', side_effect=[{'id': 7}, RuntimeError('offline')]) as request:
+                with self.assertRaises(RuntimeError):
+                    train.upload('pod', 'left', path, activate=True)
+                self.assertEqual(request.call_count, 2)
+
+    def test_conflicting_flags_refused_before_any_work(self):
+        for flags in [['--no-upload', '--select'], ['--no-upload', '--activate'], ['--select', '--activate']]:
+            with patch('sys.argv', ['train', '--pod', 'pod'] + flags), patch.object(train, 'snapshot') as snapshot:
+                with self.assertRaises(SystemExit) as error:
+                    train.main()
+                self.assertEqual(error.exception.code, 2)
+                snapshot.assert_not_called()
+
+
+class LabelTests(unittest.TestCase):
+    def test_explicit_gap_not_bridged_by_same_stage(self):
+        rows = [{'ts': t, 'watch_stage': stage} for t, stage in [(0, 'light'), (60, ''), (120, 'light'), (180, 'light')]]
+        epochs = [{'start': mid - 150} for mid in [30, 60, 90, 120, 150, 210, 240]]
+        self.assertEqual([e['start'] + 150 for e, _ in models.label(epochs, rows)], [30, 120, 150, 210])
+
+    def test_missing_sample_still_bridges_and_stage_change_does_not(self):
+        epochs = [{'start': -60}]
+        self.assertEqual(len(models.label(epochs, [{'ts': 0, 'watch_stage': 'light'}, {'ts': 120, 'watch_stage': 'light'}])), 1)
+        self.assertEqual(models.label(epochs, [{'ts': 0, 'watch_stage': 'light'}, {'ts': 120, 'watch_stage': 'deep'}]), [])
+
+class CommandTests(unittest.TestCase):
+    def test_command_defaults_candidate_and_propagates_explicit_flags(self):
+        from types import SimpleNamespace
+        for flag in [[], ['--select'], ['--activate'], ['--no-upload']]:
+            with tempfile.TemporaryDirectory() as tmp:
+                args = ['train', '--pod', 'pod', '--work', tmp, '--snapshot', 'existing.db'] + flag
+                with patch('sys.argv', args), patch.object(train, 'run', return_value=SimpleNamespace(stdout='logistic: pooled kappa 0.2\nforest: pooled kappa 0.3\n')), patch.object(train, 'upload') as upload, patch('sys.stdout', new_callable=io.StringIO):
+                    train.main()
+                if flag == ['--no-upload']:
+                    upload.assert_not_called()
+                else:
+                    upload.assert_called_once_with('pod', 'left', Path(tmp) / 'contract-B/personal-forest.json', select=flag == ['--select'], activate=flag == ['--activate'])
+
+    def test_snapshot_cleanup_and_no_scored_model(self):
+        from types import SimpleNamespace
+        with tempfile.TemporaryDirectory() as tmp:
+            db = Path(tmp) / 'biometrics.db'; db.write_text('synthetic')
+            with patch('sys.argv', ['train', '--pod', 'pod', '--work', tmp]), patch.object(train, 'snapshot', return_value=db), patch.object(train, 'run', return_value=SimpleNamespace(stdout='no models')), patch('sys.stdout', new_callable=io.StringIO):
+                with self.assertRaisesRegex(SystemExit, 'no model scored'):
+                    train.main()
+            self.assertFalse(db.exists())
+
+    def test_http_error_reports_endpoint_without_selecting(self):
+        error = train.urllib.error.HTTPError('http://pod', 400, 'bad', {}, io.BytesIO(b'invalid model'))
+        with patch.object(train.urllib.request, 'urlopen', side_effect=error):
+            with self.assertRaisesRegex(SystemExit, 'pod refused models \\(400\\): invalid model'):
+                train.request_json('pod', 'models', {})
+
+    def test_snapshot_transport_and_subprocess_wrapper(self):
+        with patch.object(train, 'run') as run:
+            self.assertEqual(train.snapshot('pod', Path('/tmp/synthetic')), Path('/tmp/synthetic/biometrics.db'))
+            self.assertEqual(run.call_count, 3)
+            self.assertIn('src.backup(dst)', run.call_args_list[0].kwargs['input'])
+        with patch.object(train.subprocess, 'run') as run:
+            train.run([Path('/tmp/synthetic-command'), 'argument'])
+            run.assert_called_once_with(['/tmp/synthetic-command', 'argument'], check=True, text=True)

scripts/train-personal-models.py

@@ -6,7 +6,7 @@
 with Apple Watch labels (scripts/sleep-stage-dataset.py --from-db), the two
 models on a feature contract with leave-one-night-out scoring and the pod's
 portable JSON export (docs/research/sleep-stage-calibration/personal-models.py
---export), then PUT /api/sleep-stages/model on the pod with the model whose
+--export), then stores and previews a candidate on the pod with the model whose
 pooled kappa is higher. Seconds on Apple silicon; nothing runs on the phone.
 
 Usage:
@@ -61,17 +61,31 @@ def kappa_of(output, kind):
     return float(match.group(1)) if match else None
 
 
-def upload(pod, side, path):
-    model = json.loads(path.read_text())
-    body = json.dumps({'side': side, 'model': model}).encode()
-    request = urllib.request.Request(f'http://{pod}:3000/api/sleep-stages/model', data=body, method='PUT',
+def request_json(pod, endpoint, body, method='POST'):
+    request = urllib.request.Request(f'http://{pod}:3000/api/sleep-stages/{endpoint}',
+                                     data=json.dumps(body).encode(), method=method,
                                      headers={'content-type': 'application/json'})
     try:
         with urllib.request.urlopen(request, timeout=30) as response:
             return json.load(response)
     except urllib.error.HTTPError as error:
         detail = error.read().decode(errors='replace')
-        sys.exit(f'pod refused the model ({error.code}): {detail[:400]}')
+        sys.exit(f'pod refused {endpoint} ({error.code}): {detail[:400]}')
+
+
+def upload(pod, side, path, *, select=False, activate=False):
+    """Store and preview a candidate. Selection requires explicit opt-in."""
+    model = json.loads(path.read_text())
+    stored = request_json(pod, 'models', {'side': side, 'model': model})
+    print(f'pod stored candidate {stored}')
+    preview = request_json(pod, 'model/preview', {'side': side, 'nights': 14, 'model': model})
+    print('Preview (training nights are in-sample):')
+    print(json.dumps(preview, indent=2))
+    if select or activate:
+        selected = request_json(pod, 'models/selected',
+                                {'side': side, 'id': stored['id'], 'active': activate}, 'PUT')
+        print(f'pod selected {selected}')
+    return stored
 
 
 def main():
@@ -83,8 +97,14 @@ def main():
     ap.add_argument('--work', default='personal-models', help='working directory (default ./personal-models)')
     ap.add_argument('--snapshot', help='an existing biometrics.db snapshot instead of taking one')
     ap.add_argument('--no-upload', action='store_true', help='train and export only')
+    ap.add_argument('--select', action='store_true', help='select the candidate, inactive (disables personal staging)')
+    ap.add_argument('--activate', action='store_true', help='select and activate the candidate after preview')
     ap.add_argument('--keep-snapshot', action='store_true', help='leave the ~75 MB database snapshot in the work dir (deleted after a run by default)')
     args = ap.parse_args()
+    if args.select and args.activate:
+        ap.error('--select and --activate cannot be combined')
+    if args.no_upload and (args.select or args.activate):
+        ap.error('--no-upload cannot be combined with --select or --activate')
 
     work = Path(args.work)
     work.mkdir(parents=True, exist_ok=True)
@@ -112,10 +132,7 @@ def main():
     print(f'chosen: {kind} (pooled kappa {scores[kind]}) -> {path}')
     if args.no_upload:
         return
-    stored = upload(args.pod, args.side, path)
-    print(f'pod stored {stored}')
-    print(f'score it there: curl -X POST http://{args.pod}:3000/api/sleep-stages/model/preview '
-          f"-H 'content-type: application/json' -d '{{\"side\":\"{args.side}\",\"nights\":14}}'")
+    upload(args.pod, args.side, path, select=args.select, activate=args.activate)
 
 
 if __name__ == '__main__':