feat(runner): minimal Pipeline SDK + BYOC hello-world E2E

[rickstaa]

Summary

The Pipeline SDK — both halves: a request/response Pipeline for batch HTTP/SSE and a LivePipeline for real-time bidirectional video/audio over BYOC trickle. Developers write a single Python class and get a containerised, BYOC-compatible, schema-described capability with /health, /docs, /openapi.json, SSE streaming, and (when complete) /stream/* real-time endpoints. No go-livepeer changes required.

Each commit on this branch ships a strictly more capable SDK than the last; test.sh stays green at every step.

Authoring surfaces

Batch / streaming HTTP — Pipeline:

from livepeer_gateway.runner import Pipeline, serve

class Sentiment(Pipeline):
    def setup(self) -> None:
        self.classifier = pipeline("sentiment-analysis", model="...")

    def predict(self, text: str) -> dict:
        return self.classifier(text)[0]

if __name__ == "__main__":
    serve(Sentiment())

For binary I/O, swap text: str for image: Base64Bytes. For streaming, return a generator from predict() — the runtime auto-detects and frames each yielded value as SSE.

Real-time A/V — LivePipeline (in progress in this PR):

from livepeer_gateway.runner import LivePipeline, serve

class GrayscaleFilter(LivePipeline):
    async def process_video(self, frame):
        # frame.frame is an av.VideoFrame; transform and return
        return frame

if __name__ == "__main__":
    serve(GrayscaleFilter())

Subclasses override any of setup, on_stream_start, process_video, process_audio, on_params_update, on_stream_stop (all default to no-op / passthrough). A subclass that overrides nothing is a valid passthrough relay.

What's in this PR

Commit-by-commit progression (each shippable on its own):

Commit	What it adds
C1 — minimal Pipeline SDK + hello-world E2E	Pipeline ABC with predict() + aiohttp serve() + first BYOC docker-compose
C2 — setup() lifecycle + sentiment example	One-time model loading hook; HuggingFace sentiment classifier example
FastAPI migration	aiohttp → FastAPI; brings /docs, /openapi.json, /redoc for free
C4 — Pydantic typed I/O via signature introspection	predict() params with type hints become a Pydantic model automatically; explicit BaseModel parameter also supported; /docs shows real fields
C5 — image upscale example (binary via Base64Bytes)	Pydantic's Base64Bytes proves the SDK handles binary I/O cleanly
C6 — /health state machine	LOADING / OK / ERROR / IDLE matching go-livepeer's HealthCheck wire format
C7 — SSE auto-detection + LLM chat example	Generator predict() → text/event-stream with [DONE] terminator; Qwen2.5-0.5B example
C8 Step 1 — LivePipeline ABC + /stream/* HTTP skeleton (landed)	New LivePipeline base class + _make_live_pipeline_app dispatch path; routes accept the orchestrator's wire contract (validated via Pydantic); streaming logic lands in subsequent commits
C8 Step 2 — trickle bytes-through on /stream/start|stop (landed)	_run_passthrough bridges subscribe → publish via existing TrickleSubscriber / TricklePublisher, segment-aligned and unmodified. /stream/start spawns the bridge as a background task on the pipeline; /stream/stop cancels and waits up to 5s for cleanup. Single-session for now (409 on double-start).
C8 Step 3 — frame decode → user transform → encode loop (planned)	Reuse existing MediaOutput / MediaPublish; introduce livepeer_gateway.runner.frames namespace (clean VideoFrame / AudioFrame aliases)
C8 Step 4 — lifecycle hooks complete (planned)	on_stream_start / on_params_update / on_stream_stop dispatch; emit_event / emit_data over events / data trickle channels; introduce _LiveSession to encapsulate per-session state
C8 Step 5 — examples/runner/live_grayscale/ end-to-end (planned)	Full BYOC compose with go-livepeer orchestrator, register_capability, test.sh exercising real stream lifecycle

Surface

Module / class	What it is
livepeer_gateway.runner.Pipeline	ABC with setup() and abstract predict(**kwargs) -> Any
livepeer_gateway.runner.LivePipeline	Base class for real-time A/V pipelines on the BYOC trickle protocol; default-passthrough hooks for process_video / process_audio
livepeer_gateway.runner.PipelineState	LOADING / OK / ERROR / IDLE enum; matches go-livepeer HealthCheck format
livepeer_gateway.runner.serve(pipeline, *, host, port)	FastAPI app + uvicorn server; dispatches on Pipeline vs LivePipeline
livepeer_gateway.runner.make_app(pipeline)	Just the FastAPI app (for tests, custom uvicorn config)
POST /predict (Pipeline)	Body validated via Pydantic from predict()'s signature; returns JSON or text/event-stream if predict() is a generator
POST /stream/start | stop | params (LivePipeline)	Real-time stream lifecycle endpoints; bodies match the orchestrator's wire contract
GET /health	HealthResponse { status: PipelineState } — orchestrator-aligned
GET /docs, GET /openapi.json, GET /redoc	Standard FastAPI surface
examples/runner/hello_world/	Smoke test — minimal Pipeline + Dockerfile + compose + register_capability + curl test.sh
examples/runner/sentiment/	setup() lifecycle + HF sentiment classifier
examples/runner/image_upscale/	Binary I/O via Pydantic Base64Bytes — Swin2SR ~2x super-resolution
examples/runner/llm/	SSE streaming via TextIteratorStreamer — Qwen2.5-0.5B-Instruct
examples/runner/live_grayscale/ (planned)	Real-time A/V pipeline E2E example via LivePipeline + go-livepeer trickle

The container's /predict and /stream/* paths match the existing go-livepeer BYOC contract verified against byoc/stream_orchestrator.go. No go-livepeer changes required.

Authoring patterns (Pipeline)

# Plain typed kwargs — Pydantic model built automatically
class Sentiment(Pipeline):
    def predict(self, text: str) -> dict: ...

# Explicit Pydantic input + output — full /docs + typed response
class Request(BaseModel):
    text: str
    threshold: float = 0.5

class Response(BaseModel):
    label: str
    score: float

class Sentiment(Pipeline):
    def predict(self, body: Request) -> Response: ...

# Streaming via generator return
class LLM(Pipeline):
    def predict(self, prompt: str) -> Iterator[ChatChunk]:
        for token in self.streamer:
            yield ChatChunk(token=token)

Test plan

Each example ships its own test.sh that prints PASS on success.

• Local Python: uv run python examples/runner/hello_world/pipeline.py → curl localhost:5000/{health,predict} returns expected JSON.
• Hello world E2E: cd examples/runner/hello_world && docker compose up -d --wait && ./test.sh → PASS. Round-trip: curl → gateway → orchestrator → SDK container.
• Sentiment: setup() loads HF model once; test.sh exercises POSITIVE / NEGATIVE cases via EXPECTED_LABEL.
• Image upscale: Base64Bytes round-trip; output asserted to be at least 2x input dimensions.
• LLM: SSE round-trip via curl -N; assertion validates token framing + [DONE] terminator.
• OpenAPI: /docs and /openapi.json render for every example with the actual field names (no additionalProp1).
• Health state machine: /health returns LOADING during setup(), OK after, ERROR on setup failure.
• LivePipeline skeleton: /stream/start|stop|params accept the orchestrator's wire contract; missing required fields → 422; /docs and /openapi.json show the new routes; batch Pipeline regression-checked.
• LivePipeline trickle lifecycle: full session lifecycle (idle stop, missing-URL → 400, double-start → 409, cancel + restart, idempotent stop) tested via curl against a local server; trickle code paths exercised end-to-end including aiohttp connection setup, segment fetch loop, cancellation propagation through finally blocks.
• LivePipeline E2E against go-livepeer: real Docker compose with live_grayscale example, full stream lifecycle (start → frames → stop) — lands with C8 Step 5.

Compose details

Each example's docker-compose.yml mirrors go-livepeer/doc/byoc.md:

• livepeer/go-livepeer:master for orchestrator + gateway (no local build prerequisite)
• On-chain mode against the free public Arbitrum RPC, pricePerUnit 0 → no real chain interaction, no funded wallet
• Drops to bare -network offchain once livepeer/go-livepeer#3906 ships in :master — TODO comments in each compose track the cleanup
• Examples that load HF models bake them into the image at build time via prepare_models.py so setup() loads from local cache in milliseconds

What's next (separate PRs, after this one merges)

Tracked in #8 — Pipeline SDK roadmap:

• C9 — livepeer push CLI + livepeer.yaml manifest
• C10 — Schema as Docker image label (org.livepeer.pipeline.schema)
• C11 — Agent-friendly docs (AGENTS.md, expanded docstrings, examples/runner/_template/)
• C12 — Migrate to monorepo with PEP 420 namespace packages (livepeer.runner.*, livepeer.client.*, livepeer.trickle.*)

Related work

• Filed upstream: livepeer/go-livepeer#3905 (issue) + #3906 (fix). When that PR ships, our compose drops the on-chain flags.
• PR feat(byoc): BYOC streaming and payments with examples #6 alignment: A future PR will swap test.sh's curl-with-base64-Livepeer-header for a Python livepeer_gateway batch caller built on PR feat(byoc): BYOC streaming and payments with examples #6's signing primitives — at that point the gateway service can be dropped from compose. TODO in test.sh tracks.
• Filed during this work: livepeer/go-livepeer#3913 — BYOC protocol cleanups surfaced while building the SDK.

🤖 Generated with Claude Code