BYOC smoke-test: CPU gateway and orchestrator (off-chain to on-chain)

mkdir byoc-cpu-test
cd byoc-cpu-test

import asyncio
import logging

from runner.live.pipelines import Pipeline, BaseParams
from runner.live.trickle import VideoFrame, VideoOutput


class PassthroughParams(BaseParams):
    """No custom parameters needed for passthrough."""
    pass


class PassthroughPipeline(Pipeline):
    """
    CPU-only passthrough pipeline for BYOC smoke testing.
    Echoes every incoming video frame directly to the output queue.
    No model loading, no GPU required.
    """

    def __init__(self):
        self.frame_queue: asyncio.Queue[VideoOutput] = asyncio.Queue()
        self._running = True

    async def initialize(self, **params):
        logging.info("PassthroughPipeline: initialized (CPU, no model loading)")

    async def put_video_frame(self, frame: VideoFrame, request_id: str):
        if self._running:
            # Pass the frame through unchanged
            await self.frame_queue.put(VideoOutput(frame, request_id))

    async def get_processed_video_frame(self) -> VideoOutput:
        return await self.frame_queue.get()

    async def update_params(self, **params):
        logging.info(f"PassthroughPipeline: params update received: {params}")

    async def stop(self):
        self._running = False
        logging.info("PassthroughPipeline: stopped")

    @classmethod
    def prepare_models(cls):
        logging.info("PassthroughPipeline: no models to prepare (CPU passthrough)")

import os
from runner.app import start_app
from runner.live.pipelines import PipelineSpec

pipeline_spec = PipelineSpec(
    name="passthrough-cpu",          # This is your model_id in go-livepeer
    pipeline_cls="pipeline:PassthroughPipeline",
    params_cls="pipeline:PassthroughParams",
    initial_params={},
)

if __name__ == "__main__":
    start_app(pipeline=pipeline_spec)

# Pin to a specific ai-runner live-base version
ARG BASE_IMAGE=livepeer/ai-runner:live-base
FROM ${BASE_IMAGE}

WORKDIR /app

# Copy pipeline files
COPY pipeline.py ./pipeline.py
COPY main.py ./main.py

# No additional dependencies needed for CPU passthrough

ENV HF_HUB_OFFLINE=1

ENTRYPOINT ["python", "main.py"]

docker build -t byoc-cpu-passthrough:latest .

docker images | grep byoc-cpu-passthrough

# Replace vX.Y.Z with the latest release from github.com/livepeer/go-livepeer/releases
curl -LO https://github.com/livepeer/go-livepeer/releases/download/vX.Y.Z/livepeer-linux-amd64.tar.gz
tar -xzf livepeer-linux-amd64.tar.gz
chmod +x livepeer livepeer_cli

docker run -d \
  --name byoc-cpu-passthrough \
  --network host \
  -p 8000:8000 \
  byoc-cpu-passthrough:latest

docker logs byoc-cpu-passthrough
# Should see: PassthroughPipeline: initialized (CPU, no model loading)

./livepeer \
  -orchestrator \
  -serviceAddr 0.0.0.0:8935 \
  -cliAddr 127.0.0.1:7935 \
  -byoc \
  -byocContainerURL http://localhost:8000 \
  -byocModelID passthrough-cpu \
  -pricePerUnit 1 \
  -network offchain \
  -datadir ./data-orchestrator

./livepeer \
  -gateway \
  -cliAddr 127.0.0.1:7936 \
  -httpAddr 0.0.0.0:8935 \
  -orchAddr http://localhost:8935 \
  -remoteSignerAddr https://signer.eliteencoder.net \
  -network offchain \
  -datadir ./data-gateway

curl http://localhost:8935/health
# Expected: {"status":"ok"}

curl http://localhost:7935/registeredOrchestrators
# Expected: JSON array with your orchestrator entry

pip install git+https://github.com/j0sh/livepeer-python-gateway.git

import asyncio
from livepeer_gateway import GatewayClient

async def main():
    client = GatewayClient(
        gateway_url="http://localhost:8935",
        model_id="passthrough-cpu",
    )

    print("Starting BYOC job...")
    async with client.start_byoc_job() as session:
        # Send a small test payload (simulated video frame as bytes)
        test_frame = b"\x00" * 1024  # 1KB of zeros — minimal CPU test payload
        await session.send(test_frame)
        print("Frame sent. Waiting for response...")

        result = await session.receive()
        print(f"Job complete. Response size: {len(result)} bytes")
        assert len(result) > 0, "Expected non-empty response from passthrough pipeline"

    print("BYOC test passed.")

asyncio.run(main())

python test_job.py

curl -X POST http://localhost:8935/live/video-to-video \
  -H "Content-Type: application/octet-stream" \
  --data-binary @/dev/urandom \
  --max-time 10

./livepeer_cli \
  -network arbitrum-one-mainnet \
  -datadir ./data-gateway \
  # Follow the CLI prompts to create a new keystore

curl -X POST YOUR_RPC_URL \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":1}'

./livepeer_cli \
  -network arbitrum-one-mainnet \
  -ethUrl YOUR_ARBITRUM_RPC \
  -datadir ./data-gateway

./livepeer_cli \
  -network arbitrum-one-mainnet \
  -ethUrl YOUR_ARBITRUM_RPC \
  -datadir ./data-gateway \
  # Select "Get node status" to see PM deposit balance

./livepeer \
  -orchestrator \
  -serviceAddr YOUR_PUBLIC_IP:8935 \
  -network arbitrum-one-mainnet \
  -ethUrl YOUR_ARBITRUM_RPC \
  -datadir ./data-orchestrator \
  -byoc \
  -byocContainerURL http://localhost:8000 \
  -byocModelID passthrough-cpu \
  -pricePerUnit 1000

./livepeer_cli \
  -network arbitrum-one-mainnet \
  -ethUrl YOUR_ARBITRUM_RPC \
  -datadir ./data-orchestrator
# Select: "Activate orchestrator"
# Set your service fee, block reward cut, and pricing

./livepeer \
  -gateway \
  -cliAddr 127.0.0.1:7936 \
  -httpAddr 0.0.0.0:8935 \
  -network arbitrum-one-mainnet \
  -ethUrl YOUR_ARBITRUM_RPC \
  -datadir ./data-gateway