Add deep pipeline integration tests for hardware suite

5 new tests in TestFullPipeline class:
- test_recording_produces_correct_json: verify frame count, timestamps, structure
- test_recording_with_audio_produces_wav: verify WAV duration matches recording
- test_record_replay_full_lifecycle: record → verify → replay → delete
- test_stop_cancels_queued_recording: immediate cancel of queued recording
- test_recording_transitions_through_phases: observe queued → countdown → recording → idle

These test the full pipeline that the unit tests can't reach — the handoff
between HTTP API and robot worker thread where the freeze bug lived.

Hardware test count: 6 → 11

tests/run_tests.py

@@ -58,6 +58,7 @@ TEST_CLASS_DESCRIPTIONS: dict[str, tuple[str, str]] = {
     "TestHardwareStartup": ("hardware", "Robot reaches idle after startup"),
     "TestHardwareRecording": ("hardware", "Record and verify motion capture"),
     "TestHardwarePlayback": ("hardware", "Playback with and without audio"),
+    "TestFullPipeline": ("hardware", "Full record → verify files → replay → delete lifecycle"),
 }
 
 

tests/test_hardware.py

@@ -16,11 +16,13 @@ Or via the test runner:
 import json
 import threading
 import time
+import wave
+from pathlib import Path
 
 import pytest
 import uvicorn
 
-from marionette.main import create_app
+from marionette.main import create_app, COUNTDOWN_SECONDS, MOTION_SAMPLE_RATE
 
 # Skip all tests in this module unless -m hardware is specified
 pytestmark = pytest.mark.hardware
@@ -258,3 +260,207 @@ class TestHardwarePlayback:
         # Verify it's gone
         state = httpx.get(f"{base_url}/api/state", timeout=5).json()
         assert not any(m["id"] == move_id for m in state["moves"])
+
+
+class TestFullPipeline:
+    """End-to-end pipeline tests: record → verify files → replay → delete.
+
+    These tests exercise the full lifecycle and check that output files
+    have the expected structure, durations, and frame counts.
+    """
+
+    def test_recording_produces_correct_json(self, base_url: str, hw_marionette):
+        """Record 3s silent, verify JSON has timestamps, frames at ~100Hz."""
+        import httpx
+
+        duration = 3.0
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": duration, "record_audio": False, "label": "pipeline-json"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+        move_id = resp.json()["move_id"]
+
+        _wait_for_mode(base_url, "idle", timeout=COUNTDOWN_SECONDS + duration + 10)
+
+        # Read the actual JSON file from disk
+        json_path = hw_marionette._dataset_dir / f"{move_id}.json"
+        assert json_path.exists(), f"JSON file not found: {json_path}"
+
+        data = json.loads(json_path.read_text())
+        timestamps = data["time"]
+        frames = data["set_target_data"]
+
+        # Frame count should be close to duration * 100Hz
+        expected_frames = int(duration * MOTION_SAMPLE_RATE)
+        assert len(frames) > expected_frames * 0.8, (
+            f"Too few frames: {len(frames)} (expected ~{expected_frames})"
+        )
+        assert len(timestamps) == len(frames)
+
+        # Timestamps should span close to the requested duration
+        actual_duration = timestamps[-1] - timestamps[0]
+        assert actual_duration > duration * 0.8, (
+            f"Recorded duration too short: {actual_duration:.2f}s (expected ~{duration}s)"
+        )
+
+        # Each frame must have head pose and antennas
+        for frame in frames[:3]:  # check first few
+            assert "head" in frame
+            assert "antennas" in frame
+
+        # Cleanup
+        httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
+
+    def test_recording_with_audio_produces_wav(self, base_url: str, hw_marionette):
+        """Record 3s with mic, verify WAV file duration matches recording."""
+        import httpx
+
+        duration = 3.0
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": duration, "record_audio": True, "label": "pipeline-audio"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+        move_id = resp.json()["move_id"]
+
+        _wait_for_mode(base_url, "idle", timeout=COUNTDOWN_SECONDS + duration + 10)
+
+        # Verify WAV exists and has reasonable duration
+        wav_path = hw_marionette._dataset_dir / f"{move_id}.wav"
+        assert wav_path.exists(), f"WAV file not found: {wav_path}"
+
+        with wave.open(str(wav_path), "rb") as wf:
+            wav_duration = wf.getnframes() / wf.getframerate()
+            assert wav_duration > duration * 0.7, (
+                f"WAV too short: {wav_duration:.2f}s (expected ~{duration}s)"
+            )
+            assert wav_duration < duration + 2.0, (
+                f"WAV too long: {wav_duration:.2f}s (expected ~{duration}s)"
+            )
+
+        # Also verify the JSON file reports has_audio
+        state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+        move = next((m for m in state["moves"] if m["id"] == move_id), None)
+        assert move is not None
+        assert move["has_audio"] is True
+
+        # Cleanup
+        httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
+
+    def test_record_replay_full_lifecycle(self, base_url: str, hw_marionette):
+        """Record → verify → replay → verify playback completes → delete."""
+        import httpx
+
+        duration = 2.0
+
+        # Step 1: Record
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": duration, "record_audio": False, "label": "lifecycle"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+        move_id = resp.json()["move_id"]
+
+        state = _wait_for_mode(base_url, "idle", timeout=COUNTDOWN_SECONDS + duration + 10)
+        assert any(m["id"] == move_id for m in state["moves"]), "Move not in list after recording"
+
+        # Step 2: Replay
+        resp = httpx.post(
+            f"{base_url}/api/play",
+            json={"move_id": move_id},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+
+        state = _wait_for_mode(base_url, "idle", timeout=duration + 15)
+        assert state["mode"] == "idle"
+
+        # Step 3: Delete and verify cleanup
+        json_path = hw_marionette._dataset_dir / f"{move_id}.json"
+        assert json_path.exists()
+
+        resp = httpx.delete(f"{base_url}/api/moves/{move_id}", timeout=5)
+        assert resp.status_code == 200
+        assert not json_path.exists(), "JSON file should be deleted"
+
+        state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+        assert not any(m["id"] == move_id for m in state["moves"])
+
+    def test_stop_cancels_queued_recording(self, base_url: str):
+        """Submit a recording, immediately stop it — verify cancel works."""
+        import httpx
+
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": 10.0, "record_audio": False, "label": "cancel-test"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+
+        # Immediately try to stop (may be queued or in countdown)
+        time.sleep(0.1)
+        resp = httpx.post(f"{base_url}/api/record/stop", timeout=5)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["stopped"] is True
+
+        # Should return to idle
+        _wait_for_mode(base_url, "idle", timeout=10)
+
+    def test_recording_transitions_through_phases(self, base_url: str):
+        """Verify the recording goes through queued → countdown → recording → idle."""
+        import httpx
+
+        duration = 3.0
+        observed_modes = set()
+
+        resp = httpx.post(
+            f"{base_url}/api/record",
+            json={"duration": duration, "record_audio": False, "label": "phases"},
+            timeout=5,
+        )
+        assert resp.status_code == 200
+
+        # Poll rapidly to observe phase transitions
+        deadline = time.time() + COUNTDOWN_SECONDS + duration + 10
+        while time.time() < deadline:
+            try:
+                state = httpx.get(f"{base_url}/api/state", timeout=2).json()
+                observed_modes.add(state["mode"])
+
+                # When in recording mode, verify timing fields are present
+                if state["mode"] == "recording":
+                    assert state["phase_start_at"] is not None, "recording phase_start_at is null"
+                    assert state["phase_end_at"] is not None, "recording phase_end_at is null"
+
+                # When in countdown, verify timing fields
+                if state["mode"] == "countdown":
+                    assert state["phase_start_at"] is not None, "countdown phase_start_at is null"
+                    assert state["phase_end_at"] is not None, "countdown phase_end_at is null"
+
+                if state["mode"] == "idle" and "phases" in (state.get("message") or ""):
+                    break  # Recording completed
+            except Exception:
+                pass
+            time.sleep(0.15)
+
+        # We should have seen at least countdown and recording phases
+        assert "countdown" in observed_modes, (
+            f"Never saw countdown mode. Observed: {observed_modes}"
+        )
+        assert "recording" in observed_modes, (
+            f"Never saw recording mode. Observed: {observed_modes}"
+        )
+        assert "idle" in observed_modes, (
+            f"Never returned to idle. Observed: {observed_modes}"
+        )
+
+        # Cleanup
+        state = httpx.get(f"{base_url}/api/state", timeout=5).json()
+        move = next((m for m in state["moves"] if "phases" in m.get("label", "")), None)
+        if move:
+            httpx.delete(f"{base_url}/api/moves/{move['id']}", timeout=5)