feat(comfystream): overlay the published image with Livepeer integration

README.md

@@ -38,7 +38,7 @@ The orchestrator is a **transparent reverse proxy**: every endpoint you expose i
 
 - **HTTP** request/response — the common case. (`hello-world`, `tiles`, `api-proxy`)
 - **HTTP + SSE** — streamed / token responses. (`vllm`)
-- **Trickle** — continuous realtime video in/out. (`echo`)
+- **Trickle** — continuous realtime video in/out. (`echo`, `comfystream`)
 - **WebSocket** — long-lived bidirectional sessions. (`realtime-transcription`)
 
 Need a transport that isn't here? [Open an issue](https://github.com/livepeer/runner-app-examples/issues).
@@ -51,6 +51,7 @@ Need a transport that isn't here? [Open an issue](https://github.com/livepeer/ru
 | [`tiles`](./tiles)                                   | Capacity fan-out — one call per tile                                                     | dynamic      | single-shot | HTTP (base64 PNG) | fixed   |
 | [`api-proxy`](./api-proxy)                           | Pass calls through to hosted APIs — the operator holds the key, one capability per model | static       | single-shot | HTTP (JPEG bytes) | fixed   |
 | [`echo`](./echo)                                     | Realtime video, transformed and echoed back                                              | dynamic      | persistent  | trickle           | hour    |
+| [`comfystream`](./comfystream)                       | ComfyUI live video + analyze; consumes the published `livepeer/comfystream` image        | dynamic      | persistent  | trickle + HTTP    | hour    |
 | [`vllm`](./vllm)                                     | Drop-in OpenAI API; the client stays unmodified                                          | static       | single-shot | HTTP + SSE        | hour    |
 | [`realtime-transcription`](./realtime-transcription) | Audio up, transcripts back, on one socket                                                | dynamic      | persistent  | WebSocket         | hour    |
 
@@ -62,7 +63,7 @@ This set stays **minimal and curated**: it covers each value of the axes above (
 
 How the app attaches to the orchestrator:
 
-- **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`, `realtime-transcription`)
+- **Dynamic** — the app self-registers via the SDK (`register_runner`) and heartbeats; the orchestrator drops it when heartbeats stop. Best for apps that come and go. (`hello-world`, `echo`, `comfystream`, `realtime-transcription`)
 - **Static** — the orchestrator is configured with the app's URL in a `runners.json` and health-polls it; the app needs no SDK. Best for fixed, long-running deployments. (`vllm`, `api-proxy`)
 
 The arrow flips: dynamic, the app announces itself; static, the orchestrator is told about a passive app:
@@ -92,15 +93,15 @@ Clients read it off the discovered runner, whose `raw` holds that runner's disco
 
 Chosen _at_ registration (above); **defaults to `persistent`**, set on both `register_runner(...)` and in `runners.json`. The examples set it explicitly.
 
-- **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `realtime-transcription`)
+- **Persistent** — a held-open session the client reserves and releases, billed per second of wall-clock (or once, with fixed pricing). Best for realtime / streaming. (`echo`, `comfystream`, `realtime-transcription`)
 - **Single-shot** — one request in, one response out; the orchestrator reserves a session per call and releases it when the response returns, so the client manages no session at all. Best for batch / request-response. With metered pricing the call pays for as long as it runs, so the work need not be short. (`hello-world`, `tiles`, `api-proxy`, `vllm`)
 
 ## Calling your app
 
 The client side depends on the runner's mode:
 
 - **Single-shot** — **discover → call**: find the app via `runner_selector`, then one `call_runner`. The orchestrator reserves a session for the call and releases it when the response returns; on the paid path `call_runner` answers the 402 payment challenge inline. (`hello-world`, `tiles`, `api-proxy`, `vllm`)
-- **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `realtime-transcription`)
+- **Persistent** — **discover → reserve → call → release**: reserve a session (`reserve_session`), call it — `call_runner`, streamed frames, or a WebSocket, depending on transport — then release it (`stop_runner_session`), which settles payment on-chain. (`echo`, `comfystream`, `realtime-transcription`)
 
 Each example's `client.py` shows its exact calls: grep `# Livepeer:` to find them.
 

comfystream/.env.example

@@ -0,0 +1,33 @@
+# Copy to .env (gitignored) and fill in. Never commit secrets.
+# Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only.
+
+NETWORK=arbitrum-one-mainnet
+ETH_RPC_URL=https://arb1.arbitrum.io/rpc
+
+# Signer (payer): needs an on-chain deposit + reserve.
+SIGNER_KEYSTORE_DIR=/absolute/path/to/signer-keystore
+SIGNER_ETH_ACCT=0xYourSignerAddress
+SIGNER_ETH_PASSWORD=your-signer-keystore-password
+
+# Orchestrator operating key (split-key): needs ETH for gas to redeem tickets.
+ORCH_KEYSTORE_DIR=/absolute/path/to/operator-keystore
+ORCH_ETH_ACCT=0xYourOperatorAddress
+ORCH_ETH_PASSWORD=your-operator-keystore-password
+# Registered orch = ticket recipient (-ethOrchAddr); empty = use the operating key.
+ORCH_ONCHAIN_ADDR=0xYourRegisteredOrchestrator
+
+# Runner price (on-chain): USD per hour, metered per second while the client
+# holds the session. Keep under ~0.67: the signer signs at most 100 tickets
+# per payment, and the demo orchestrator runs -ticketEV=1e10 (fee / ticketEV).
+PRICE=0.10
+# Signer's max-price cap (payer side), per billing unit. Metered here, so the
+# unit is one second and must exceed PRICE / 3600 (0.000111USD is ~0.40/hour).
+MAX_PRICE_PER_UNIT=0.000111USD
+
+# Optional: pin a GPU and host model/storage dirs when using compose.existing.yml.
+# COMFYSTREAM_GPU_UUID=GPU-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+# COMFYSTREAM_MODELS_DIR=/livepeer/ai/data/models
+# COMFYSTREAM_STORAGE_DIR=/livepeer/ai/data
+# LIVEPEER_ORCH_URL=https://127.0.0.1:8935
+# LIVEPEER_ORCH_SECRET=abcdef
+# LIVEPEER_RUNNER_URL=http://127.0.0.1:8991

comfystream/Dockerfile

@@ -0,0 +1,23 @@
+# Overlay: published ComfyStream image + this folder's live-runner integration.
+# The comfystream package and ComfyUI workspace already live in the base image.
+ARG BASE_IMAGE=livepeer/comfystream:latest
+FROM ${BASE_IMAGE}
+
+WORKDIR /app
+
+# livepeer-gateway's generated pb2 needs protobuf >= 6.31.1. ComfyUI pins
+# protobuf<5; the live-runner path still runs with protobuf 6.
+RUN /bin/bash -lc 'source /workspace/miniconda3/etc/profile.d/conda.sh \
+  && conda activate comfystream \
+  && pip install --no-cache-dir \
+    "protobuf>=6.31.1" \
+    "livepeer-gateway>=1.0.0" \
+    av'
+
+COPY runner.py /app/runner.py
+COPY entrypoint.sh /app/entrypoint.sh
+RUN chmod +x /app/entrypoint.sh
+
+EXPOSE 8991
+
+ENTRYPOINT ["/app/entrypoint.sh"]

comfystream/README.md

@@ -0,0 +1,98 @@
+# ComfyStream app (workflow-driven video + analyze)
+
+A ComfyUI live-video app on the Livepeer network. It consumes the published
+[`livepeer/comfystream`](https://hub.docker.com/r/livepeer/comfystream) image —
+the `comfystream` package and ComfyUI workspace stay in that image. This folder
+is only the Livepeer integration: dynamic registration, trickle channels, and
+the analyze / start_stream HTTP surface.
+
+|              |                                      |
+| ------------ | ------------------------------------ |
+| App id       | `livepeer-example/comfystream`       |
+| Runner mode  | persistent (held-open session)       |
+| Registration | dynamic (self-registers via the SDK) |
+| Transport    | trickle + HTTP (`/analyze`)          |
+| Pricing      | hour (metered per second)            |
+| Port         | 8991                                 |
+
+Prerequisites (Docker, NVIDIA GPU, `uv`, the
+[`livepeer-gateway` SDK](https://pypi.org/project/livepeer-gateway/)) and the
+shared setup are in the [repo README](../README.md). You also need the
+`livepeer/comfystream` image (pulled as the Docker build base).
+
+## How it's wired
+
+ComfyStream is **installed, not vendored**. The Dockerfile starts `FROM
+livepeer/comfystream:latest` and adds `livepeer-gateway` into that image's conda
+env. `runner.py` imports `comfystream.Pipeline` from the package already in the
+image and registers with the orchestrator.
+
+The app is **dynamically registered**: it self-registers via `register_runner`
+([runner.py](runner.py)) and exposes:
+
+- `POST /analyze` — video-in → text-out (trickle `in` + JSONL `text`)
+- `POST /start_stream` — live trickle video (optional text)
+- `POST /update_stream` — mid-session workflow / resolution change
+- `GET /text` — buffered text for the active session
+- `GET /healthz`
+
+The client calls it with `reserve_session` → `post_json` / `MediaPublish` →
+`stop_runner_session` ([client.py](client.py)). Grep `# Livepeer:` in either
+file to see the exact calls.
+
+This is the same persistent + trickle shape as [`echo`](../echo), plus an HTTP
+analyze surface driven by a ComfyUI API-format workflow JSON.
+
+## Run offchain (free)
+
+Start the stack and confirm the runner registered. The first build pulls the
+ComfyStream image and can take a while.
+
+```sh
+docker compose up -d --build
+curl -sk https://localhost:8935/discovery | jq '.[].runners[].app'   # confirm livepeer-example/comfystream registered
+```
+
+Any video works; this makes a short one if you need it:
+
+```sh
+ffmpeg -f lavfi -i testsrc=size=512x512:rate=30 -t 5 -c:v libx264 -preset ultrafast -pix_fmt yuv420p sample.mp4
+uv run client.py sample.mp4 --discovery https://localhost:8935/discovery
+```
+
+`client.py` defaults to [workflows/analyze-stub-api.json](workflows/analyze-stub-api.json)
+(video-in → a fixed text token). Pass `--workflow path/to/workflow.json` for a
+real ComfyUI graph, and `--mode stream` for `start_stream`.
+
+Stop the stack with `docker compose down`.
+
+## Attach to an existing orchestrator
+
+Same image and runner, no local go-livepeer. Point at the orch you already run
+(env vars in `.env` / `.env.example`):
+
+```sh
+docker compose -f compose.existing.yml up -d --build
+uv run client.py sample.mp4 --discovery "$LIVEPEER_ORCH_URL/discovery"
+```
+
+## Run on-chain (paid)
+
+Layer `compose.onchain.yml` to run the orchestrator on-chain with a remote
+signer paying for the session. This example uses **metered pricing**: `PRICE`
+is USD per hour, billed per second for as long as the client holds the session.
+For the required RPC and wallets see
+[On-chain (paid) setup](../README.md#on-chain-paid-setup) in the repo README.
+
+```sh
+cp .env.example .env   # fill in RPC, network, keystore paths, accounts, pricing
+docker compose -f compose.yml -f compose.onchain.yml up -d --build
+uv run client.py sample.mp4 \
+  --discovery https://localhost:8935/discovery \
+  --signer http://localhost:7936
+docker compose -f compose.yml -f compose.onchain.yml down
+```
+
+A metered session pays **more than once**. Use a clip of at least a few tens of
+seconds if you want to see repeated payments in
+`docker compose logs -f orchestrator`.

comfystream/client.py

@@ -0,0 +1,195 @@
+#!/usr/bin/env python3
+"""comfystream client: reserve a session, drive analyze / start_stream, release.
+
+Publishes video frames into the runner's trickle `in` channel. Analyze collects
+text via GET /text; start_stream can also swap the workflow mid-session.
+
+Livepeer integration (grep `# Livepeer:`):
+  1. reserve_session()
+  2. post_json / MediaPublish through session.app_url (orch injects session headers)
+  3. stop_runner_session()
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+from contextlib import suppress
+from pathlib import Path
+from typing import Any
+
+import av
+from livepeer_gateway.errors import LivepeerGatewayError
+from livepeer_gateway.http import get_json, post_json
+from livepeer_gateway.live_runner import stop_runner_session
+from livepeer_gateway.media_publish import MediaPublish
+from livepeer_gateway.selection import reserve_session
+
+APP_ID = "livepeer-example/comfystream"
+DEFAULT_DISCOVERY = "https://localhost:8935/discovery"
+DEFAULT_WORKFLOW = (
+    Path(__file__).resolve().parent / "workflows" / "analyze-stub-api.json"
+)
+log = logging.getLogger("comfystream-client")
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Run the proxied ComfyStream Live Runner demo."
+    )
+    parser.add_argument("input", help="Input video file.")
+    parser.add_argument("--discovery", default=DEFAULT_DISCOVERY)
+    parser.add_argument(
+        "--signer",
+        default="",
+        help="Remote signer URL for on-chain path.",
+    )
+    parser.add_argument(
+        "--workflow",
+        default=str(DEFAULT_WORKFLOW),
+        help="ComfyUI API-format workflow JSON (default: analyze stub).",
+    )
+    parser.add_argument(
+        "--mode",
+        choices=("analyze", "stream", "both"),
+        default="analyze",
+        help="Which surface to exercise (default: analyze).",
+    )
+    parser.add_argument("--max-frames", type=int, default=30)
+    parser.add_argument("--width", type=int, default=512)
+    parser.add_argument("--height", type=int, default=512)
+    parser.add_argument(
+        "--update-workflow",
+        default="",
+        help="Optional second workflow JSON for update_stream.",
+    )
+    return parser.parse_args()
+
+
+def _load_workflow(path: str) -> Any:
+    return json.loads(Path(path).read_text(encoding="utf-8"))
+
+
+async def _publish_frames(
+    publish_url: str,
+    input_path: str,
+    *,
+    max_frames: int,
+) -> None:
+    publisher = MediaPublish(publish_url)
+    try:
+        container = av.open(input_path)
+        sent = 0
+        for frame in container.decode(video=0):
+            await publisher.write_frame(frame)
+            sent += 1
+            if max_frames and sent >= max_frames:
+                break
+        container.close()
+        log.info("published %d frames to %s", sent, publish_url)
+    finally:
+        await publisher.close()
+
+
+async def _run_analyze(args: argparse.Namespace, workflow: Any) -> None:
+    session = await reserve_session(  # Livepeer: 1
+        discovery_url=args.discovery,
+        app=APP_ID,
+        signer_url=args.signer.strip() or None,
+    )
+    try:
+        async with session:
+            data = await post_json(  # Livepeer: 2
+                f"{session.app_url.rstrip('/')}/analyze",
+                {
+                    "prompts": workflow,
+                    "width": args.width,
+                    "height": args.height,
+                },
+                timeout=120.0,
+            )
+            log.info("analyze started: %s", data)
+            await _publish_frames(data["in"], args.input, max_frames=args.max_frames)
+            await asyncio.sleep(2.0)
+            texts = await get_json(
+                f"{session.app_url.rstrip('/')}/text",
+                timeout=30.0,
+            )
+            log.info("analyze texts: %s", texts)
+    except LivepeerGatewayError as exc:
+        raise SystemExit(f"ERROR: {exc}") from exc
+    finally:
+        with suppress(Exception):
+            await stop_runner_session(session)  # Livepeer: 3
+
+
+async def _run_stream(args: argparse.Namespace, workflow: Any) -> None:
+    session = await reserve_session(  # Livepeer: 1
+        discovery_url=args.discovery,
+        app=APP_ID,
+        signer_url=args.signer.strip() or None,
+    )
+    try:
+        async with session:
+            data = await post_json(  # Livepeer: 2
+                f"{session.app_url.rstrip('/')}/start_stream",
+                {
+                    "prompts": workflow,
+                    "width": args.width,
+                    "height": args.height,
+                },
+                timeout=120.0,
+            )
+            log.info("stream started: %s", data)
+
+            publish = asyncio.create_task(
+                _publish_frames(data["in"], args.input, max_frames=args.max_frames)
+            )
+            if args.update_workflow:
+                await asyncio.sleep(1.0)
+                update = _load_workflow(args.update_workflow)
+                updated = await post_json(
+                    f"{session.app_url.rstrip('/')}/update_stream",
+                    {"prompts": update},
+                    timeout=60.0,
+                )
+                log.info("update_stream: %s", updated)
+            await publish
+            await asyncio.sleep(1.0)
+    except LivepeerGatewayError as exc:
+        raise SystemExit(f"ERROR: {exc}") from exc
+    finally:
+        with suppress(Exception):
+            await stop_runner_session(session)  # Livepeer: 3
+        log.info("stream session stopped")
+
+
+async def _amain() -> int:
+    args = _parse_args()
+    input_path = Path(args.input).expanduser()
+    if not input_path.exists():
+        raise SystemExit(f"input file does not exist: {input_path}")
+    args.input = str(input_path)
+    workflow = _load_workflow(args.workflow)
+    if args.mode in ("analyze", "both"):
+        await _run_analyze(args, workflow)
+    if args.mode in ("stream", "both"):
+        await _run_stream(args, workflow)
+    return 0
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(message)s",
+    )
+    try:
+        raise SystemExit(asyncio.run(_amain()))
+    except KeyboardInterrupt:
+        raise SystemExit(130) from None
+
+
+if __name__ == "__main__":
+    main()