ci: catch drift in the restated orchestrator commands

.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())