Add critical hardware tests for real-world breakage scenarios

- TestStopDuringPlayback: stop mid-playback and during goto-start-pose
- TestStopDuringRecording: stop mid-recording, verify partial save
- TestMultipleRecordPlayCycles: 5 back-to-back record/play cycles
- TestPlaybackWithCorruptFile: deleted/truncated JSON recovery

These cover the most common user interactions that were previously
untested on real hardware: interrupting playback, cancelling recordings,
repeated usage sessions, and filesystem inconsistencies.

Total: 95 unit + 44 E2E + 37 hardware = 176 tests.

TESTING.md

@@ -196,6 +196,10 @@ pytest tests/ --browser chromium
 | TestRecordingRoundTrip | 2 | Record → playback fidelity and timing verification |
 | TestAntennaAndBodyYaw | 2 | Verify antenna and body_yaw data in recordings |
 | TestPlaybackAntennas | 1 | Synthetic antenna oscillation playback |
+| TestStopDuringPlayback | 2 | Stop mid-playback and during goto-start-pose |
+| TestStopDuringRecording | 1 | Stop mid-recording, verify partial save |
+| TestMultipleRecordPlayCycles | 1 | 5 back-to-back record/play cycles |
+| TestPlaybackWithCorruptFile | 2 | Deleted/corrupt JSON during playback |
 | TestHardwareAudio | 3 | Audio recording and playback (may skip on mic issues) |
 
 ## View the matrix without running tests

tests/run_tests.py

@@ -83,6 +83,10 @@ TEST_CLASS_DESCRIPTIONS: dict[str, tuple[str, str]] = {
     "TestRecordingRoundTrip": ("hardware", "Record → playback fidelity and timing verification"),
     "TestAntennaAndBodyYaw": ("hardware", "Verify antenna and body_yaw data in recordings"),
     "TestPlaybackAntennas": ("hardware", "Synthetic antenna oscillation playback"),
+    "TestStopDuringPlayback": ("hardware", "Stop mid-playback and during goto-start-pose"),
+    "TestStopDuringRecording": ("hardware", "Stop mid-recording, verify partial save"),
+    "TestMultipleRecordPlayCycles": ("hardware", "5 back-to-back record/play cycles"),
+    "TestPlaybackWithCorruptFile": ("hardware", "Deleted/corrupt JSON during playback"),
 }
 
 

tests/test_hardware.py

@@ -1211,6 +1211,273 @@ class TestPlaybackAntennas:
         httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
 
 
+class TestStopDuringPlayback:
+    """Stop playback mid-stream — the #1 user-initiated interruption."""
+
+    def test_stop_during_playback_returns_to_idle(self, base_url: str, hw_marionette):
+        """Start playback, stop mid-way, verify robot returns to idle gracefully."""
+        import httpx
+
+        _ensure_idle(base_url)
+
+        # Record a 5s move so we have time to stop mid-playback
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": 5.0, "record_audio": False, "label": "stop-play-test"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+        move_id = resp.json()["move_id"]
+        _wait_for_mode(base_url, "idle", timeout=COUNTDOWN_SECONDS + 5 + 10)
+
+        # Start playback
+        resp = httpx.post(
+            f"{base_url}/api/play",
+            json={"move_id": move_id},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+
+        # Wait until we're actually in playing mode
+        _wait_for_mode(base_url, "playing", timeout=15)
+        # Let it play for 1-2 seconds
+        time.sleep(1.5)
+
+        # Stop mid-playback
+        resp = httpx.post(f"{base_url}/api/play/stop", timeout=5)
+        assert resp.status_code == 200
+
+        # Should return to idle
+        state = _wait_for_mode(base_url, "idle", timeout=10)
+        assert state["mode"] == "idle"
+
+        # Cleanup
+        httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
+
+    def test_stop_during_goto_start_pose(self, base_url: str, hw_marionette):
+        """Stop playback during the initial goto-start-pose transition."""
+        import httpx
+        import numpy as np
+
+        _ensure_idle(base_url)
+
+        # Create synthetic move with head far from neutral — long goto transition
+        move_id = _create_synthetic_move(
+            hw_marionette,
+            label="stop-goto-test",
+            duration=5.0,
+            trajectory_fn=lambda t: (0.0, 0.0, 0.4),  # constant yaw offset
+        )
+
+        # Start playback — goto_pose_scaled will take time
+        resp = httpx.post(
+            f"{base_url}/api/play",
+            json={"move_id": move_id},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+
+        # Stop immediately (likely still in goto phase)
+        time.sleep(0.3)
+        resp = httpx.post(f"{base_url}/api/play/stop", timeout=5)
+        assert resp.status_code == 200
+
+        state = _wait_for_mode(base_url, "idle", timeout=15)
+        assert state["mode"] == "idle"
+
+        # Cleanup
+        httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
+
+
+class TestStopDuringRecording:
+    """Stop recording mid-capture — verify partial data is saved correctly."""
+
+    def test_stop_mid_recording_saves_partial(self, base_url: str, hw_marionette):
+        """Record 10s, stop after 2-3s, verify partial JSON is saved with valid data."""
+        import httpx
+
+        _ensure_idle(base_url)
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": 10.0, "record_audio": False, "label": "stop-partial"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+        move_id = resp.json()["move_id"]
+
+        # Wait for actual recording mode (not countdown)
+        _wait_for_mode(base_url, "recording", timeout=COUNTDOWN_SECONDS + 5)
+        # Let it record for 2 seconds
+        time.sleep(2.0)
+
+        # Stop recording
+        resp = httpx.post(f"{base_url}/api/record/stop", timeout=5)
+        assert resp.status_code == 200
+
+        state = _wait_for_mode(base_url, "idle", timeout=10)
+        assert state["mode"] == "idle"
+
+        # Verify partial recording was saved with reasonable data
+        json_path = hw_marionette._dataset_dir / f"{move_id}.json"
+        assert json_path.exists(), "Partial recording should be saved"
+
+        data = json.loads(json_path.read_text())
+        timestamps = data["time"]
+        frames = data["set_target_data"]
+
+        # Should have ~2s worth of data (not 10s)
+        assert len(frames) > 100, f"Expected ~200 frames for 2s, got {len(frames)}"
+        assert len(frames) < 500, f"Expected <5s of data, got {len(frames)} frames"
+        assert len(timestamps) == len(frames)
+
+        # Verify frames have valid structure
+        for frame in frames[:5]:
+            assert "head" in frame
+            assert "antennas" in frame
+
+        print(f"\nPartial recording: {len(frames)} frames in {timestamps[-1]:.2f}s")
+
+        # Cleanup
+        httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
+
+
+class TestMultipleRecordPlayCycles:
+    """Back-to-back record/play cycles — the real-world usage pattern.
+
+    Users record several takes in a row, playing them back between takes.
+    Motor state drift and resource leaks show up after multiple cycles.
+    """
+
+    def test_five_record_play_cycles(self, base_url: str, hw_marionette):
+        """Run 5 record→play cycles, verify each completes and robot stays healthy."""
+        import httpx
+
+        move_ids = []
+        for i in range(5):
+            _ensure_idle(base_url)
+
+            # Record 1.5s
+            resp = httpx.post(
+                f"{base_url}/api/record",
+                json={"duration": 1.5, "record_audio": False, "label": f"cycle-{i}"},
+                timeout=5,
+            )
+            assert resp.status_code == 200, f"Cycle {i}: record failed"
+            move_id = resp.json()["move_id"]
+            move_ids.append(move_id)
+            _wait_for_mode(base_url, "idle", timeout=COUNTDOWN_SECONDS + 1.5 + 10)
+
+            # Verify recording exists
+            state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+            assert any(m["id"] == move_id for m in state["moves"]), (
+                f"Cycle {i}: move {move_id} not in list"
+            )
+
+            # Play it back
+            resp = httpx.post(
+                f"{base_url}/api/play",
+                json={"move_id": move_id},
+                timeout=5,
+            )
+            assert resp.status_code == 200, f"Cycle {i}: play failed"
+            _wait_for_mode(base_url, "idle", timeout=1.5 + 15)
+
+            print(f"  Cycle {i+1}/5 complete")
+
+        # Verify all 5 moves still exist after all cycles
+        state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+        for mid in move_ids:
+            assert any(m["id"] == mid for m in state["moves"]), (
+                f"Move {mid} disappeared after cycles"
+            )
+
+        # Cleanup
+        for mid in move_ids:
+            httpx.delete(f"{base_url}/api/moves/{mid}", timeout=5)
+
+
+class TestPlaybackWithCorruptFile:
+    """Playback when the JSON file is missing or corrupt.
+
+    Users may delete files from the dataset folder while the app is running.
+    """
+
+    def test_playback_deleted_file_returns_error(self, base_url: str, hw_marionette):
+        """Delete a move's JSON mid-session, try to play — should fail gracefully."""
+        import httpx
+
+        _ensure_idle(base_url)
+
+        # Record a move
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": 1.5, "record_audio": False, "label": "corrupt-test"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+        move_id = resp.json()["move_id"]
+        _wait_for_mode(base_url, "idle", timeout=COUNTDOWN_SECONDS + 1.5 + 10)
+
+        # Delete the JSON file behind the app's back
+        json_path = hw_marionette._dataset_dir / f"{move_id}.json"
+        assert json_path.exists()
+        json_path.unlink()
+
+        # Refresh so the app picks up the deletion
+        hw_marionette._refresh_recordings()
+
+        # Try to play — should get 404 (move no longer exists)
+        resp = httpx.post(
+            f"{base_url}/api/play",
+            json={"move_id": move_id},
+            timeout=5,
+        )
+        assert resp.status_code in (404, 409), (
+            f"Expected 404/409 for deleted move, got {resp.status_code}"
+        )
+
+        state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+        assert state["mode"] in ("idle", "error")
+
+    def test_playback_truncated_json_recovers(self, base_url: str, hw_marionette):
+        """Write a truncated JSON, try to play — should fail without crashing."""
+        import httpx
+
+        _ensure_idle(base_url)
+
+        # Create a corrupt JSON file
+        move_id = "corrupt-truncated"
+        json_path = hw_marionette._dataset_dir / f"{move_id}.json"
+        json_path.write_text('{"time": [0.0, 0.01], "set_target_data": [{"head', encoding="utf-8")
+        hw_marionette._refresh_recordings()
+
+        # Try to play — should fail gracefully (corrupt JSON can't be loaded)
+        resp = httpx.post(
+            f"{base_url}/api/play",
+            json={"move_id": move_id},
+            timeout=5,
+        )
+
+        # The move may not parse during refresh (skipped) or fail during load
+        if resp.status_code == 200:
+            # If it was accepted, it should recover to idle/error
+            time.sleep(3)
+            state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+            assert state["mode"] in ("idle", "error"), (
+                f"Server stuck after corrupt playback: mode={state['mode']}"
+            )
+        else:
+            # 404 is fine — refresh skipped the corrupt file
+            assert resp.status_code in (404, 409, 500)
+
+        _ensure_idle(base_url)
+
+        # Cleanup
+        if json_path.exists():
+            json_path.unlink()
+        hw_marionette._refresh_recordings()
+
+
 class TestHardwareAudio:
     """Audio recording/playback tests — run last.