Add smoke test to verify full playback pipeline on robot

smoke_test_on_robot.py starts Marionette with the real dataset root,
connects to the robot, waits for startup, lists existing recordings,
picks one with audio, and plays it back via the API. Verifies the
entire flow works locally on the robot (motors + audio).

run_smoke_on_robot.py launches it from the laptop via SSH.

tests/run_smoke_on_robot.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""Run the smoke test on the robot via SSH.
+
+Syncs code, then runs smoke_test_on_robot.py on the robot.
+Streams output in real-time so you can see if the robot moves
+and audio plays.
+
+Usage:
+    cd marionette
+    python tests/run_smoke_on_robot.py
+    python tests/run_smoke_on_robot.py --host 192.168.1.42
+"""
+
+import argparse
+import subprocess
+import sys
+from pathlib import Path
+
+# Reuse sync logic from run_on_robot
+sys.path.insert(0, str(Path(__file__).parent))
+from run_on_robot import sync_to_robot, ensure_deps, ssh_cmd, REMOTE_DIR, ROBOT_PYTHON
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Run smoke test on the Reachy Mini robot via SSH.",
+    )
+    parser.add_argument("--host", default="reachy-mini.local")
+    parser.add_argument("--user", default="pollen")
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    # Sync code
+    print(f"Syncing code to {args.user}@{args.host}...")
+    ok = sync_to_robot(args.host, args.user, dry_run=args.dry_run)
+    if not ok:
+        print("Failed to sync.")
+        sys.exit(1)
+    if args.dry_run:
+        print("Dry run — not running test.")
+        sys.exit(0)
+
+    # Ensure deps
+    if not ensure_deps(args.host, args.user):
+        print("Failed to install deps.")
+        sys.exit(1)
+
+    # Run smoke test
+    remote_cmd = (
+        f"cd {REMOTE_DIR} && "
+        f"PYTHONPATH={REMOTE_DIR} "
+        f"{ROBOT_PYTHON} tests/smoke_test_on_robot.py"
+    )
+
+    print(f"\n{'=' * 60}")
+    print(f"Running smoke test on {args.user}@{args.host}")
+    print(f"{'=' * 60}\n")
+
+    proc = subprocess.Popen(
+        ssh_cmd(args.host, args.user, remote_cmd),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+    )
+
+    for line in proc.stdout:
+        print(line, end="")
+
+    proc.wait()
+    sys.exit(proc.returncode)
+
+
+if __name__ == "__main__":
+    main()

tests/smoke_test_on_robot.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""Smoke test: start Marionette on the robot and play back an existing move.
+
+This script is meant to be run ON the robot (after rsync) to verify
+that the full pipeline works locally: server starts, finds existing
+recordings, plays one back with audio and motion.
+
+Usage (on the robot):
+    cd /tmp/marionette_test
+    PYTHONPATH=/tmp/marionette_test /venvs/apps_venv/bin/python tests/smoke_test_on_robot.py
+
+Or from the laptop via run_on_robot.py flow:
+    python tests/run_smoke_on_robot.py
+"""
+
+import json
+import sys
+import threading
+import time
+
+import httpx
+import uvicorn
+
+from marionette.main import create_app
+
+PORT = 18044
+BASE_URL = f"http://127.0.0.1:{PORT}"
+STARTUP_TIMEOUT = 45  # seconds — startup animation can be slow
+
+
+def wait_for_mode(target_mode: str, timeout: float) -> dict:
+    deadline = time.time() + timeout
+    last = {}
+    while time.time() < deadline:
+        try:
+            resp = httpx.get(f"{BASE_URL}/api/state", timeout=5)
+            last = resp.json()
+            if last.get("mode") == target_mode:
+                return last
+        except Exception:
+            pass
+        time.sleep(0.3)
+    raise TimeoutError(
+        f"Timed out waiting for mode={target_mode!r} "
+        f"(last mode={last.get('mode')!r})"
+    )
+
+
+def main():
+    # Use the REAL default dataset root (not a temp dir) so we see
+    # existing recordings on the robot.
+    # Pass registry_path=None and dataset_root=None to use defaults.
+    print("Creating Marionette app with default (real) dataset root...")
+    app, marionette = create_app()
+
+    # Connect to robot
+    print("Connecting to robot...")
+    from reachy_mini import ReachyMini
+    reachy = ReachyMini()
+    print("  Connected!")
+
+    stop_event = threading.Event()
+
+    # Start the run loop (startup animation, then idle)
+    run_thread = threading.Thread(
+        target=marionette.run, args=(reachy, stop_event), daemon=True,
+    )
+    run_thread.start()
+
+    # Start web server
+    config = uvicorn.Config(app, host="127.0.0.1", port=PORT, log_level="warning")
+    server = uvicorn.Server(config)
+    srv_thread = threading.Thread(target=server.run, daemon=True)
+    srv_thread.start()
+
+    # Wait for startup animation to finish
+    print(f"Waiting for startup animation (up to {STARTUP_TIMEOUT}s)...")
+    try:
+        wait_for_mode("idle", STARTUP_TIMEOUT)
+    except TimeoutError as e:
+        print(f"  FAILED: {e}")
+        stop_event.set()
+        server.should_exit = True
+        sys.exit(1)
+    print("  Startup complete — server is idle.")
+
+    # List available moves
+    state = httpx.get(f"{BASE_URL}/api/state", timeout=5).json()
+    moves = state.get("moves", [])
+    print(f"\nFound {len(moves)} moves:")
+    for i, m in enumerate(moves):
+        audio = "with audio" if m.get("has_audio") else "no audio"
+        print(f"  [{i}] {m['id']} — {m.get('duration', '?'):.1f}s, {audio}")
+
+    if not moves:
+        print("\nNo moves found! Make sure the robot has existing recordings.")
+        print(f"Dataset path: {state['config']['active_dataset_path']}")
+        stop_event.set()
+        server.should_exit = True
+        sys.exit(1)
+
+    # Pick the first move that has audio, or fall back to any move
+    move = next((m for m in moves if m.get("has_audio")), moves[0])
+    print(f"\nPlaying: {move['id']} ({move.get('duration', '?'):.1f}s, "
+          f"{'with' if move.get('has_audio') else 'no'} audio)")
+
+    # Play it!
+    resp = httpx.post(
+        f"{BASE_URL}/api/play",
+        json={"move_id": move["id"]},
+        timeout=5,
+    )
+    if resp.status_code != 200:
+        print(f"  FAILED to start playback: {resp.status_code} {resp.text}")
+        stop_event.set()
+        server.should_exit = True
+        sys.exit(1)
+
+    print("  Playback started...")
+
+    # Wait for playback to finish, printing status
+    t0 = time.time()
+    timeout = move.get("duration", 30) + 20
+    while time.time() - t0 < timeout:
+        try:
+            st = httpx.get(f"{BASE_URL}/api/state", timeout=5).json()
+            mode = st["mode"]
+            if mode == "idle":
+                elapsed = time.time() - t0
+                print(f"  Playback finished in {elapsed:.1f}s")
+                break
+            elif mode == "playing":
+                # Show progress
+                pass
+        except Exception:
+            pass
+        time.sleep(0.5)
+    else:
+        print("  WARNING: playback did not return to idle in time")
+
+    # Final state
+    final = httpx.get(f"{BASE_URL}/api/state", timeout=5).json()
+    print(f"\nFinal mode: {final['mode']}")
+    print("\nSMOKE TEST PASSED" if final["mode"] == "idle" else "\nSMOKE TEST FAILED")
+
+    # Shutdown
+    stop_event.set()
+    server.should_exit = True
+    run_thread.join(timeout=5)
+
+    sys.exit(0 if final["mode"] == "idle" else 1)
+
+
+if __name__ == "__main__":
+    main()