feat(echo): add a robot mode that transforms the audio (#58)

echo dropped audio in three places, so trickle looked video-only: the
client decoded video=0, the runner returned None for non-video frames,
and only a video track was published. A stream with sound came back
silent with no explanation.

robot ring-modulates the audio and leaves the video alone, which makes
the round trip audible. Ring modulation rather than a pitch shift
because it preserves the sample count, so audio stays in sync without
resampling. The output track is pinned to 48 kHz since opus rejects
44.1, which every consumer source produces.

Only robot publishes an audio track: the container opens once every
declared track has a first frame, so declaring audio for a silent input
would stall the video.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: rickstaa

.pre-commit-config.yaml

@@ -24,3 +24,15 @@ repos:
       - id: end-of-file-fixer
       - id: check-yaml
       - id: check-merge-conflict
+
+  - repo: local
+    hooks:
+      - id: orchestrator-drift
+        # api-proxy and vllm restate the shared orchestrator command because they
+        # add -liveRunnerConfig and `extends` replaces a list. Keep the copies honest.
+        name: orchestrator command drift
+        entry: python3 scripts/check_orchestrator_drift.py
+        language: python
+        additional_dependencies: [pyyaml]
+        pass_filenames: false
+        files: ^(compose\.orchestrator\.yml|compose\.onchain\.yml|[^/]+/compose(\.onchain)?\.yml)$

echo/README.md

@@ -1,6 +1,6 @@
 # Echo app (trickle realtime video)
 
-A realtime video app on the Livepeer network: it receives a live video stream over **trickle** channels, optionally transforms each frame (gray / invert / blur), and echoes it back. This is the **live/stateful** path — continuous media over trickle, not request/response — so the app embeds the SDK and self-registers (dynamic).
+A realtime video app on the Livepeer network: it receives a live video stream over **trickle** channels, optionally transforms each frame (gray / invert / blur) or the audio (robot), and echoes it back. This is the **live/stateful** path — continuous media over trickle, not request/response — so the app embeds the SDK and self-registers (dynamic).
 
 |              |                                      |
 | ------------ | ------------------------------------ |
@@ -73,10 +73,27 @@ Swap `/dev/video0` for your node. If that size/format isn't supported, list the
 
 The `ffplay` low-delay flags (`-fflags nobuffer -flags low_delay -framedrop`) keep the preview close to realtime; drop them and it buffers.
 
-- `--mode` picks the transform: `echo` (passthrough, the default), `gray`, `invert`, or `blur`. Use `--mode blur` on any command above to see the echo visibly transform the stream.
+- `--mode` picks the transform: `echo` (passthrough, the default), `gray`, `invert`, `blur`, or `robot`. Use `--mode blur` on any command above to see the echo visibly transform the stream.
+- `robot` ring-modulates the audio and leaves the video alone. It is the only mode that publishes an audio track, since a declared track that never gets a frame stalls the stream.
 - `blur` sweeps the radius `0 -> max -> 0` live (driving `/update`); `--blur-period N` sets the seconds per sweep cycle (default 2; larger is slower). `gray`/`invert` are static.
 - `--radius N` sets the initial blur strength, `--max-frames N` stops early.
 
+**Hearing `robot`** — every command above is video-only, so `robot` would refuse them. Record yourself with a microphone (`arecord -l` lists capture devices), then play both files:
+
+```sh
+ffmpeg -f v4l2 -input_format mjpeg -video_size 1280x720 -framerate 30 -i /dev/video0 \
+  -f alsa -i plughw:1,0 -filter_complex "[1:a]aresample=async=1:first_pts=0[a]" \
+  -map 0:v -map "[a]" -fps_mode cfr -t 10 \
+  -c:v libx264 -preset ultrafast -pix_fmt yuv420p -g 30 -c:a aac -ar 48000 -f mpegts -y me.ts
+
+uv run client.py --mode robot --output me-robot.ts me.ts
+ffplay -autoexit me.ts && ffplay -autoexit me-robot.ts   # you, then you ring-modulated
+```
+
+To hear it live instead, keep the same capture and swap the tail for `--output - -` piped into `ffplay -fflags nobuffer -i -`. Expect 2 to 4 seconds of lag, since trickle publishes in 2s segments, and wear headphones or the mic re-records the playback.
+
+The camera and the audio device are separate clocks, so `aresample` and `-fps_mode cfr` align them; without both the publisher fails at the first segment boundary. On a multi-input interface add `-channels 6` and pick one input with `pan=mono|c0=c0`.
+
 Stop the stack with `docker compose down`.
 
 ## Run on-chain (paid)

echo/client.py

@@ -32,15 +32,20 @@
 from livepeer_gateway.errors import LivepeerGatewayError
 from livepeer_gateway.live_runner import stop_runner_session
 from livepeer_gateway.media_output import MediaOutput
-from livepeer_gateway.media_publish import MediaPublish
+from livepeer_gateway.media_publish import (
+    AudioOutputConfig,
+    MediaPublish,
+    MediaPublishConfig,
+    VideoOutputConfig,
+)
 from livepeer_gateway.http import post_json
 from livepeer_gateway.selection import reserve_session
 
 DEFAULT_DISCOVERY = "https://localhost:8935/discovery"
 APP_ID = "livepeer-example/echo"
 DEFAULT_OUTPUT = "echo-out.ts"
 MAX_BLUR_RADIUS = 100
-MODES = ("echo", "gray", "invert", "blur")
+MODES = ("echo", "gray", "invert", "blur", "robot")
 
 log = logging.getLogger("echo-client")
 
@@ -79,7 +84,8 @@ def _parse_args() -> argparse.Namespace:
         choices=MODES,
         default="echo",
         help=(
-            "Transform the runner applies: echo (passthrough), gray, invert, or blur. "
+            "Transform the runner applies: echo (passthrough), gray, invert, blur, "
+            "or robot (ring-modulates the audio). "
             "blur sweeps the radius; the rest are static."
         ),
     )
@@ -117,7 +123,20 @@ async def _publish_video(
             raise LivepeerGatewayError(
                 f"No video stream found in input: {input_source}"
             )
-        publisher = MediaPublish(publish_url)  # Livepeer: 2 (publish frames)
+        # Only robot touches audio, so only robot publishes an audio track: the
+        # container waits for a first frame on every track it declares.
+        send_audio = mode == "robot"
+        if send_audio and not input_.streams.audio:
+            raise LivepeerGatewayError(
+                f"robot needs audio, but the input has none: {input_source}"
+            )
+        tracks: list[VideoOutputConfig | AudioOutputConfig] = [VideoOutputConfig()]
+        if send_audio:
+            # Pinned: opus rejects 44.1 kHz, so let MediaPublish resample to 48.
+            tracks.append(AudioOutputConfig(sample_rate=48000))
+        publisher = MediaPublish(  # Livepeer: 2 (publish frames)
+            publish_url, config=MediaPublishConfig(tracks=tracks)
+        )
         prev_pts_time: float | None = None
         prev_wall: float | None = None
         next_update_pts_time: float | None = None
@@ -126,9 +145,18 @@ async def _publish_video(
         # blur sweeps 0->max->0 (2*MAX steps); spread one full cycle over blur_period.
         update_interval = blur_period / (2 * MAX_BLUR_RADIUS)
 
+        video_index = 0
         try:
-            for index, frame in enumerate(input_.decode(video=0), start=1):
-                if max_frames > 0 and index > max_frames:
+            # decode() yields both streams interleaved; without audio, stay on the
+            # video stream alone. Pacing and the blur sweep run off video frames only.
+            frames = input_.decode() if send_audio else input_.decode(video=0)
+            for frame in frames:
+                if not isinstance(frame, av.VideoFrame):
+                    await publisher.write_frame(frame)
+                    continue
+
+                video_index += 1
+                if max_frames > 0 and video_index > max_frames:
                     break
                 current_pts_time = None
                 if frame.pts is not None and frame.time_base is not None:
@@ -209,7 +237,13 @@ async def main() -> None:
         async with session:
             echo = await post_json(
                 f"{session.app_url.rstrip('/')}/echo",
-                {"radius": args.radius, "mode": args.mode},
+                # robot is the mode that transforms audio, so it is also the one
+                # that asks the runner for an audio track.
+                {
+                    "radius": args.radius,
+                    "mode": args.mode,
+                    "audio": args.mode == "robot",
+                },
             )
             in_url = _channel_url(echo, "in")
             out_url = _channel_url(echo, "out")

echo/pyproject.toml

@@ -6,6 +6,7 @@ requires-python = ">=3.12"
 dependencies = [
     "av",               # client + runner: decode/encode video frames
     "opencv-python-headless",  # runner: gray/invert/blur transforms
+    "numpy",            # runner: robot ring modulation
     "aiohttp",
     "livepeer-gateway",
 ]

echo/runner.py

@@ -25,18 +25,27 @@
 from typing import Any
 
 import av
+import numpy as np
 from aiohttp import web
 
 from livepeer_gateway.live_runner import register_runner
 from livepeer_gateway.media_decode import AudioDecodedMediaFrame, VideoDecodedMediaFrame
 from livepeer_gateway.media_output import MediaOutput
-from livepeer_gateway.media_publish import MediaPublish
+from livepeer_gateway.media_publish import (
+    AudioOutputConfig,
+    MediaPublish,
+    MediaPublishConfig,
+    VideoOutputConfig,
+)
 
 log = logging.getLogger("echo")
 
 DEFAULT_HOST = "127.0.0.1"
 DEFAULT_PORT = 8989
-MODES = frozenset({"echo", "gray", "invert", "blur"})
+MODES = frozenset({"echo", "gray", "invert", "blur", "robot"})
+# "robot" multiplies each sample by a sine at this frequency; the sample count is
+# unchanged, so audio stays in sync with video.
+ROBOT_HZ = 220.0
 
 state: EchoSession | None = None
 
@@ -121,15 +130,37 @@ def _odd_kernel(radius: int) -> int:
     return min(kernel, 99)
 
 
+def _robot_audio(frame: av.AudioFrame) -> av.AudioFrame:
+    # sample[i] *= sin(2*pi*ROBOT_HZ*t[i]). The carrier phase comes from the frame's
+    # own timestamp, so it stays continuous across frames (no clicks) without keeping
+    # state, and |carrier| <= 1 means it cannot clip.
+    samples = frame.to_ndarray()
+    t0 = float(frame.pts * frame.time_base) if frame.pts is not None else 0.0
+    t = t0 + np.arange(samples.shape[-1], dtype=np.float32) / frame.sample_rate
+    carrier = np.sin(2.0 * np.pi * ROBOT_HZ * t).astype(np.float32)
+    out = av.AudioFrame.from_ndarray(
+        (samples.astype(np.float32) * carrier).astype(samples.dtype),
+        format=frame.format.name,
+        layout=frame.layout.name,
+    )
+    out.sample_rate = frame.sample_rate
+    out.pts = frame.pts
+    out.time_base = frame.time_base
+    return out
+
+
 def _transform_frame(
     decoded: AudioDecodedMediaFrame | VideoDecodedMediaFrame,
     mode: ModeState,
-) -> av.VideoFrame | None:
+) -> av.VideoFrame | av.AudioFrame | None:
+    frame = decoded.frame
+    if decoded.kind == "audio":
+        # Audio rides along untouched; only "robot" transforms it.
+        return _robot_audio(frame) if mode.mode == "robot" else frame
     if decoded.kind != "video":
         return None
 
-    frame = decoded.frame
-    if mode.mode == "echo":
+    if mode.mode in ("echo", "robot"):  # robot changes audio only
         return frame
 
     import cv2
@@ -175,9 +206,21 @@ async def _handle_echo(request: web.Request) -> web.Response:
         )
 
     # for production apps, handle errors
-    mode = _parse_mode(json.loads(await request.read()))
+    payload = json.loads(await request.read())
+    mode = _parse_mode(payload)
+    # Tracks are declared upfront and the container waits for a first frame on each,
+    # so only declare audio when the client says it is sending some.
+    tracks: list[VideoOutputConfig | AudioOutputConfig] = [VideoOutputConfig()]
+    send_audio = payload.get("audio", False)
+    if not isinstance(send_audio, bool):
+        raise web.HTTPBadRequest(text="audio must be a boolean")
+    if send_audio:
+        tracks.append(AudioOutputConfig())
     # internal_url: runner-reachable address (same as the public url on a shared net).
-    publisher = MediaPublish(by_name["out"].get("internal_url", by_name["out"]["url"]))
+    publisher = MediaPublish(
+        by_name["out"].get("internal_url", by_name["out"]["url"]),
+        config=MediaPublishConfig(tracks=tracks),
+    )
 
     async def _on_frame(decoded) -> None:
         frame = _transform_frame(decoded, mode)

scripts/check_orchestrator_drift.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""Fail if an example's restated orchestrator command has drifted from the shared one.
+
+`compose.orchestrator.yml` and `compose.onchain.yml` define the orchestrator once and
+examples pull it in with `extends`. Static runners cannot: they need an extra
+`-liveRunnerConfig` flag, and `extends` replaces a command list rather than appending
+to it, so they restate the whole thing. This checks those copies still match.
+
+Run directly, or via pre-commit. Exits non-zero and prints what differs.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import yaml
+
+# Flags a copy may add to the shared command. Anything else is drift.
+ALLOWED_EXTRA = {"-liveRunnerConfig"}
+
+SHARED = {
+    "compose.yml": Path("compose.orchestrator.yml"),
+    "compose.onchain.yml": Path("compose.onchain.yml"),
+}
+
+
+def _command(path: Path) -> list[str] | None:
+    doc = yaml.safe_load(path.read_text()) or {}
+    service = (doc.get("services") or {}).get("orchestrator") or {}
+    command = service.get("command")
+    return command if isinstance(command, list) else None
+
+
+def _flag(item: str) -> str:
+    return str(item).split("=", 1)[0]
+
+
+def main() -> int:
+    root = Path(__file__).resolve().parent.parent
+    tracked = subprocess.run(
+        ["git", "ls-files", "*/compose.yml", "*/compose.onchain.yml"],
+        cwd=root,
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.split()
+
+    problems: list[str] = []
+    checked = 0
+    for rel in sorted(tracked):
+        path = root / rel
+        copy = _command(path)
+        if copy is None:  # uses `extends` alone, nothing to drift
+            continue
+        shared_path = root / SHARED[Path(rel).name]
+        shared = _command(shared_path) or []
+        checked += 1
+
+        shared_by_flag = {_flag(i): i for i in shared}
+        copy_by_flag = {_flag(i): i for i in copy}
+
+        for flag, item in shared_by_flag.items():
+            if flag not in copy_by_flag:
+                problems.append(f"{rel}: missing {item!r} (in {shared_path.name})")
+            elif copy_by_flag[flag] != item:
+                problems.append(
+                    f"{rel}: {flag} is {copy_by_flag[flag]!r}, "
+                    f"{shared_path.name} has {item!r}"
+                )
+        for flag, item in copy_by_flag.items():
+            if flag not in shared_by_flag and flag not in ALLOWED_EXTRA:
+                problems.append(f"{rel}: unexpected {item!r} not in {shared_path.name}")
+
+    if problems:
+        print("orchestrator command drift:\n")
+        for p in problems:
+            print(f"  {p}")
+        print(
+            "\nThese examples restate the shared command because they add "
+            f"{sorted(ALLOWED_EXTRA)} and `extends` cannot append to a list. "
+            "Re-copy the shared command, or widen ALLOWED_EXTRA if the new flag "
+            "is deliberate."
+        )
+        return 1
+
+    print(f"orchestrator command: {checked} restated copies match the shared files")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())