diff --git a/CHANGELOG.md b/CHANGELOG.md index 67074c8..3b40583 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +### Removed + +- BYOC (bring your own container) inference and training jobs: the + `submit_byoc_job`, `submit_training_job`, `refresh_training_payment`, + `get_training_status`, `wait_for_training`, and `list_capabilities` + functions, and the `ByocJobRequest`, `ByocJobResponse`, + `ByocTrainingRequest`, `ByocTrainingResponse`, and `ByocTrainingStatus` + types. BYOC is being removed from go-livepeer in + [livepeer/go-livepeer#4064](https://github.com/livepeer/go-livepeer/pull/4064). + ## [1.0.0] - 2026-08-11 The first stable release of the Livepeer Python SDK. @@ -31,4 +43,5 @@ The first stable release of the Livepeer Python SDK. `grpcio>=1.76.0` and `protobuf>=6.31.1`. - Completed the package metadata and documented installation from PyPI. +[Unreleased]: https://github.com/livepeer/livepeer-python-gateway/compare/v1.0.0...HEAD [1.0.0]: https://github.com/livepeer/livepeer-python-gateway/releases/tag/v1.0.0 diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index bc23a23..b3eaac3 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -2,19 +2,6 @@ from .channel_reader import ChannelEventCallback, ChannelReader, JSONLReader from .channel_writer import ChannelWriter, JSONLWriter from .control import Control, ControlConfig, ControlMode -from .byoc import ( - ByocJobRequest, - ByocJobResponse, - ByocTrainingRequest, - ByocTrainingResponse, - ByocTrainingStatus, - submit_byoc_job, - submit_training_job, - refresh_training_payment, - get_training_status, - wait_for_training, - list_capabilities, -) from .errors import LivepeerHTTPError, LivepeerGatewayError, NoOrchestratorAvailableError, NoRunnerAvailableError, PaymentError from .events import Events from .media_publish import ( @@ -85,17 +72,11 @@ "Control", "ControlConfig", "ControlMode", - "ByocJobRequest", - "ByocJobResponse", - "ByocTrainingRequest", - "ByocTrainingResponse", - "ByocTrainingStatus", "ChannelWriter", "CapabilityId", "build_capabilities", "discover_orchestrators", "discover_runners", - "get_training_status", "get_orch_info", "LiveVideoToVideo", "LiveRunnerCallResult", @@ -150,12 +131,9 @@ "create_trickle_channels", "register_runner", "remove_trickle_channels", - "refresh_training_payment", "start_lv2v", "start_scope", "stop_runner_session", - "submit_byoc_job", - "submit_training_job", "TricklePublishError", "TricklePublisher", "TricklePublisherStats", @@ -166,6 +144,4 @@ "TrickleSubscriber", "TrickleSubscriberStats", "VideoDecodedMediaFrame", - "wait_for_training", - "list_capabilities", ] diff --git a/src/livepeer_gateway/byoc.py b/src/livepeer_gateway/byoc.py deleted file mode 100644 index 272d668..0000000 --- a/src/livepeer_gateway/byoc.py +++ /dev/null @@ -1,973 +0,0 @@ -""" -BYOC (Bring Your Own Capability) job submission for the Livepeer network. - -Provides a simple synchronous API to submit inference requests (image generation, -video generation, music, etc.) to a Livepeer BYOC orchestrator. - -On-chain usage (with signer for payment tickets): - from livepeer_gateway.byoc import submit_byoc_job, ByocJobRequest - - result = submit_byoc_job( - req=ByocJobRequest(capability="recraft-v4", payload={"prompt": "a dragon"}), - orch_url="https://byoc-orch.daydream.monster:8935", - signer_url="https://signer.daydream.live", - signer_headers={"Authorization": "Bearer sk_..."}, - ) - print(result.image_url) - -Offchain usage (no payment, for testing): - result = submit_byoc_job( - req=ByocJobRequest(capability="nano-banana", payload={"prompt": "a cat"}), - orch_url="https://localhost:8935", - ) - - # With discovery: - result = submit_byoc_job( - discovery_url="https://discovery.example.com", - req=ByocJobRequest(capability="recraft-v4", payload={"prompt": "sunset"}), - ) -""" - -from __future__ import annotations - -import base64 -import json -import logging -import ssl -import uuid -from dataclasses import dataclass, field -from typing import Any, Optional, Sequence -from urllib.error import HTTPError, URLError -from urllib.parse import urlparse -from urllib.request import Request, urlopen - -from .orchestrator import _http_origin, discover_orchestrators -from .errors import LivepeerGatewayError, NoOrchestratorAvailableError, OrchestratorRejection - -_LOG = logging.getLogger(__name__) - -# Reusable SSL context (skip verification for self-signed certs) -_ssl_ctx = ssl.create_default_context() -_ssl_ctx.check_hostname = False -_ssl_ctx.verify_mode = ssl.CERT_NONE - - -# --------------------------------------------------------------------------- -# Data classes -# --------------------------------------------------------------------------- - -@dataclass(frozen=True) -class ByocJobRequest: - """A BYOC job request to submit to the network.""" - - capability: str - """Capability name (e.g. 'nano-banana', 'recraft-v4', 'ltx-t2v-23').""" - - payload: dict[str, Any] = field(default_factory=dict) - """Request body payload (sent as JSON).""" - - timeout_seconds: int = 300 - """Maximum time the orchestrator should wait for the worker response.""" - - job_id: Optional[str] = None - """Optional job ID. Auto-generated if not provided.""" - - parameters: Optional[dict[str, Any]] = None - """Optional job parameters (orchestrator filtering, video ingress/egress).""" - - -@dataclass -class ByocJobResponse: - """Response from a BYOC job submission.""" - - data: Any - """Parsed JSON response body from the orchestrator/worker.""" - - status_code: int = 200 - """HTTP status code.""" - - headers: dict[str, str] = field(default_factory=dict) - """Response headers (includes Livepeer-Balance, etc.).""" - - orchestrator_url: Optional[str] = None - """The orchestrator URL that processed this request.""" - - raw_body: bytes = b"" - """Raw response body bytes.""" - - @property - def balance(self) -> Optional[str]: - return self.headers.get("Livepeer-Balance") or self.headers.get("livepeer-balance") - - @property - def images(self) -> list[dict]: - """Extract images from response (convenience).""" - if isinstance(self.data, dict): - return self.data.get("images", []) - return [] - - @property - def image_url(self) -> Optional[str]: - """Extract first image URL from response.""" - for img in self.images: - if "url" in img: - return img["url"] - if isinstance(self.data, dict): - return self.data.get("image_url") or self.data.get("url") - return None - - @property - def video_url(self) -> Optional[str]: - """Extract video URL from response.""" - if not isinstance(self.data, dict): - return None - if "video" in self.data: - vid = self.data["video"] - return vid.get("url") if isinstance(vid, dict) else vid - return self.data.get("video_url") or self.data.get("url") - - @property - def audio_url(self) -> Optional[str]: - """Extract audio URL from response.""" - if not isinstance(self.data, dict): - return None - if "audio" in self.data: - aud = self.data["audio"] - return aud.get("url") if isinstance(aud, dict) else aud - if "audio_file" in self.data: - af = self.data["audio_file"] - return af.get("url") if isinstance(af, dict) else af - return self.data.get("url") - - -# --------------------------------------------------------------------------- -# Header building -# --------------------------------------------------------------------------- - -def _create_byoc_payment( - *, - orch_origin: str, - capability: str, - livepeer_hdr: str, - signer_url: str, - signer_headers: Optional[dict[str, str]] = None, - timeout: float = 30.0, -) -> dict[str, str]: - """ - Create on-chain payment tickets for a BYOC job. - - Flow: - 1. Get OrchestratorInfo via gRPC (same as LV2V) — contains ticket params + price - 2. Generate payment via signer (/generate-live-payment) - 3. Return headers to include in the job request - - Returns dict with Livepeer-Payment and Livepeer-Segment headers. - """ - from .orch_info import get_orch_info - - # Step 1: Get OrchestratorInfo via gRPC (port 8935) - # The BYOC orch_origin is on :8936 (HTTP), but gRPC is on :8935. - # Derive the gRPC URL from the HTTP origin. - parsed = urlparse(orch_origin) - grpc_url = f"https://{parsed.hostname}:8935" - - info = get_orch_info( - grpc_url, - signer_url=signer_url, - signer_headers=signer_headers, - ) - - # Check if orch has a price set — if price is 0, skip payment - if info.HasField("ticket_params"): - tp = info.ticket_params - if not tp.face_value or tp.face_value == b'\x00': - _LOG.info("BYOC orch ticket face_value=0, skipping payment") - return {} - else: - _LOG.info("BYOC orch has no ticket_params, skipping payment") - return {} - - # Step 2: Generate payment via signer - orch_info_b64 = base64.b64encode(info.SerializeToString()).decode("ascii") - - signer_origin = _http_origin(signer_url) - payment_url = f"{signer_origin}/generate-live-payment" - payment_body = json.dumps({ - "orchestrator": orch_info_b64, - "type": "lv2v", - "capability": capability, - }).encode("utf-8") - payment_headers = { - "Content-Type": "application/json", - "Livepeer-Capability": capability, - } - if signer_headers: - payment_headers.update(signer_headers) - - payment_req = Request(payment_url, data=payment_body, headers=payment_headers, method="POST") - try: - with urlopen(payment_req, timeout=timeout) as resp: - payment_data = json.loads(resp.read()) - except HTTPError as e: - body = e.read().decode("utf-8", errors="replace")[:200] - raise LivepeerGatewayError(f"BYOC payment generation failed: HTTP {e.code}: {body}") from e - - result = {} - if payment_data.get("payment"): - result["Livepeer-Payment"] = payment_data["payment"] - if payment_data.get("segCreds"): - result["Livepeer-Segment"] = payment_data["segCreds"] - - # Distinguish "signer returned empty payment" (bug) from "orch - # face_value=0" (noop, returned at line 186-190 above as `{}`). - # If we reached this point, the orch wanted a payment but the signer - # gave us nothing — raise rather than silently return `{}` so the - # caller sees a real error. - if not result: - raise LivepeerGatewayError( - "BYOC payment generation: signer returned 200 but empty " - "payment/segCreds. This is a signer bug or misconfiguration." - ) - - _LOG.info("BYOC payment tickets generated for %s", orch_origin) - return result - - -def _sign_byoc_job( - signer_url: str, - signer_headers: Optional[dict[str, str]], - job_id: str, - capability: str, - request_json: str, - parameters_json: str, - timeout_seconds: int, -) -> dict: - """Call signer /sign-byoc-job to get sender + signature for the BYOC header.""" - from .orchestrator import _http_origin - - url = f"{_http_origin(signer_url)}/sign-byoc-job" - payload = { - "id": job_id, - "capability": capability, - "request": request_json, - "parameters": parameters_json, - "timeout_seconds": timeout_seconds, - } - headers = {"Content-Type": "application/json"} - if signer_headers: - headers.update(signer_headers) - - req = Request(url, data=json.dumps(payload).encode(), headers=headers, method="POST") - try: - with urlopen(req, timeout=30.0) as resp: - return json.loads(resp.read()) - except HTTPError as e: - body = e.read().decode("utf-8", errors="replace")[:200] - raise LivepeerGatewayError(f"sign-byoc-job failed: HTTP {e.code}: {body}") from e - - -def _build_livepeer_header( - req: ByocJobRequest, - job_id: str, - sender: str = "", - sig: str = "", -) -> str: - """Build the base64-encoded Livepeer job request header.""" - request_json = json.dumps(req.payload) - parameters_json = json.dumps(req.parameters) if req.parameters else "" - job_request = { - "id": job_id, - "request": request_json, - "capability": req.capability, - "timeout_seconds": req.timeout_seconds, - } - if parameters_json: - job_request["parameters"] = parameters_json - if sender: - job_request["sender"] = sender - if sig: - job_request["sig"] = sig - return base64.b64encode(json.dumps(job_request).encode()).decode() - - -# --------------------------------------------------------------------------- -# Core API -# --------------------------------------------------------------------------- - -def submit_byoc_job( - req: ByocJobRequest, - *, - orch_url: Optional[Sequence[str] | str] = None, - discovery_url: Optional[str] = None, - signer_url: Optional[str] = None, - signer_headers: Optional[dict[str, str]] = None, - discovery_headers: Optional[dict[str, str]] = None, - timeout: Optional[float] = None, -) -> ByocJobResponse: - """ - Submit a BYOC job request to the Livepeer network. - - Discovers an orchestrator (or uses explicit orch_url), builds the Livepeer - header, and POSTs the request to /process/request/{capability}. - - Args: - req: The job request (capability, payload, timeout). - orch_url: Direct orchestrator URL(s). Highest priority. - discovery_url: Discovery endpoint to find orchestrators. - signer_url: Remote signer URL (also used for discovery fallback). - signer_headers: Headers for signer requests. - discovery_headers: Headers for discovery requests. - timeout: HTTP request timeout in seconds. Defaults to req.timeout_seconds. - - Returns: - ByocJobResponse with parsed result data. - - Raises: - NoOrchestratorAvailableError: No orchestrator could process the request. - LivepeerGatewayError: Network or protocol error. - """ - job_id = req.job_id or str(uuid.uuid4()) - http_timeout = timeout or req.timeout_seconds - - # Discover orchestrators - orch_list = _resolve_orchestrators( - orch_url=orch_url, - discovery_url=discovery_url, - signer_url=signer_url, - signer_headers=signer_headers, - discovery_headers=discovery_headers, - ) - - _LOG.info("BYOC job %s: capability=%s, orchestrators=%s", job_id, req.capability, orch_list) - - # Sign the job request if signer is available (on-chain) - sender = "" - sig = "" - if signer_url: - try: - request_json = json.dumps(req.payload) - parameters_json = json.dumps(req.parameters) if req.parameters else "" - sign_resp = _sign_byoc_job( - signer_url=signer_url, - signer_headers=signer_headers, - job_id=job_id, - capability=req.capability, - request_json=request_json, - parameters_json=parameters_json, - timeout_seconds=req.timeout_seconds, - ) - sender = sign_resp.get("sender", "") - sig = sign_resp.get("signature", "") - _LOG.info("BYOC job %s: signed by sender=%s", job_id, sender[:12] + "..." if sender else "none") - except Exception as e: - _LOG.warning("BYOC job %s: signing failed: %s", job_id, e) - - # Build headers - livepeer_hdr = _build_livepeer_header(req, job_id, sender=sender, sig=sig) - body = json.dumps(req.payload).encode("utf-8") - - # Try each orchestrator - rejections: list[OrchestratorRejection] = [] - - for orch in orch_list: - orch_origin = _http_origin(orch) - url = f"{orch_origin}/process/request/{req.capability}" - - headers = { - "Content-Type": "application/json", - "Livepeer": livepeer_hdr, - "Livepeer-Capability": req.capability, - } - - # On-chain payment: get token from orch, create payment via signer - if signer_url: - try: - payment_headers = _create_byoc_payment( - orch_origin=orch_origin, - capability=req.capability, - livepeer_hdr=livepeer_hdr, - signer_url=signer_url, - signer_headers=signer_headers, - timeout=http_timeout, - ) - headers.update(payment_headers) - _LOG.info("BYOC job %s: payment tickets created for %s", job_id, orch_origin) - except Exception as e: - _LOG.warning("BYOC job %s: payment creation failed for %s: %s", job_id, orch_origin, e) - rejections.append(OrchestratorRejection(url=orch_origin, reason=f"payment failed: {e}")) - continue - - http_req = Request(url, data=body, headers=headers, method="POST") - - _LOG.info("BYOC job %s: trying orchestrator %s", job_id, orch_origin) - - try: - with urlopen(http_req, timeout=http_timeout, context=_ssl_ctx) as resp: - raw_body = resp.read() - resp_headers = {k: v for k, v in resp.headers.items()} - - try: - data = json.loads(raw_body.decode("utf-8")) - except (json.JSONDecodeError, UnicodeDecodeError): - data = raw_body - - return ByocJobResponse( - data=data, - status_code=resp.status, - headers=resp_headers, - orchestrator_url=orch_origin, - raw_body=raw_body, - ) - - except HTTPError as e: - err_body = "" - try: - err_body = e.read().decode("utf-8", errors="replace")[:500] - except Exception: - pass - reason = f"HTTP {e.code}: {err_body}" - _LOG.warning("BYOC job %s: orchestrator %s rejected: %s", job_id, orch_origin, reason) - - # Non-retryable (4xx except 408/429) - if 400 <= e.code < 500 and e.code not in (408, 429): - raise LivepeerGatewayError( - f"BYOC job rejected by orchestrator {orch_origin}: {reason}" - ) from e - - rejections.append(OrchestratorRejection(url=orch_origin, reason=reason)) - - except (URLError, ConnectionRefusedError, TimeoutError, OSError) as e: - reason = f"{type(e).__name__}: {e}" - _LOG.warning("BYOC job %s: orchestrator %s unreachable: %s", job_id, orch_origin, reason) - rejections.append(OrchestratorRejection(url=orch_origin, reason=reason)) - - reasons = "; ".join(r.reason for r in rejections) if rejections else "no orchestrators configured" - raise NoOrchestratorAvailableError( - f"No orchestrator available for capability '{req.capability}': {reasons}", - rejections=rejections, - ) - - -def list_capabilities( - adapter_url: str, - *, - timeout: float = 10.0, -) -> list[dict[str, Any]]: - """ - List capabilities registered on an adapter. - - Args: - adapter_url: Base URL of the inference adapter (e.g. http://34.134.195.88:9090). - timeout: HTTP timeout. - - Returns: - List of capability dicts with 'name', 'model_id', 'capacity' keys. - """ - url = f"{adapter_url.rstrip('/')}/capabilities" - http_req = Request(url, headers={"Accept": "application/json"}) - - try: - with urlopen(http_req, timeout=timeout) as resp: - data = json.loads(resp.read().decode("utf-8")) - return data.get("capabilities", []) - except Exception as e: - _LOG.warning("Failed to list capabilities from %s: %s", adapter_url, e) - raise LivepeerGatewayError(f"Failed to list capabilities: {e}") from e - - -# --------------------------------------------------------------------------- -# Training API -# --------------------------------------------------------------------------- - -@dataclass(frozen=True) -class ByocTrainingRequest: - """A BYOC training job request.""" - - capability: str - """Capability name for the training job.""" - - model_id: str - """fal.ai model ID for training (e.g. 'fal-ai/flux-lora-fast-training').""" - - params: dict[str, Any] = field(default_factory=dict) - """Training parameters (images_data_url, trigger_word, steps, etc.).""" - - timeout_seconds: int = 300 - """Timeout for the initial submit request (not the training itself).""" - - callback_url: Optional[str] = None - """Optional webhook URL for completion notification.""" - - -@dataclass -class ByocTrainingResponse: - """Response from a BYOC training job submission.""" - - job_id: str - """Unique job ID for status polling.""" - - status: str = "submitted" - """Current status: submitted, running, completed, failed, cancelled.""" - - orchestrator_url: Optional[str] = None - """The orchestrator handling this job.""" - - status_url: Optional[str] = None - """Full URL to poll for status.""" - - data: Optional[dict] = None - """Raw response data.""" - - @property - def is_done(self) -> bool: - return self.status in ("completed", "failed", "cancelled") - - -@dataclass -class ByocTrainingStatus: - """Status of a training job.""" - - job_id: str - status: str - progress: int = 0 - result: Optional[dict] = None - error: Optional[str] = None - model_id: Optional[str] = None - created_at: Optional[float] = None - updated_at: Optional[float] = None - cost: Optional[str] = None - """Total cost charged so far (wei).""" - balance: Optional[str] = None - """Remaining sender balance (wei).""" - - @property - def is_done(self) -> bool: - return self.status in ("completed", "failed", "cancelled") - - @property - def lora_url(self) -> Optional[str]: - """Extract LoRA weights URL from completed result.""" - if not self.result: - return None - # fal.ai returns diffusers_lora_file.url - lora_file = self.result.get("diffusers_lora_file") - if isinstance(lora_file, dict): - return lora_file.get("url") - return self.result.get("lora_url") - - @property - def config_url(self) -> Optional[str]: - """Extract config file URL from completed result.""" - if not self.result: - return None - config_file = self.result.get("config_file") - if isinstance(config_file, dict): - return config_file.get("url") - return None - - -def submit_training_job( - req: ByocTrainingRequest, - *, - orch_url: Optional[Sequence[str] | str] = None, - discovery_url: Optional[str] = None, - signer_url: Optional[str] = None, - signer_headers: Optional[dict[str, str]] = None, - discovery_headers: Optional[dict[str, str]] = None, - timeout: Optional[float] = None, -) -> ByocTrainingResponse: - """ - Submit an async training job to the Livepeer BYOC network. - - Returns immediately with a job_id that can be polled for status. - - Args: - req: Training request (capability, model_id, params). - orch_url: Direct orchestrator URL(s). - discovery_url: Discovery endpoint. - timeout: HTTP timeout for the submit request. - - Returns: - ByocTrainingResponse with job_id and status_url. - """ - job_id = str(uuid.uuid4()) - http_timeout = timeout or req.timeout_seconds - - orch_list = _resolve_orchestrators( - orch_url=orch_url, - discovery_url=discovery_url, - signer_url=signer_url, - signer_headers=signer_headers, - discovery_headers=discovery_headers, - ) - - # Build the Livepeer header (reuse existing infrastructure) - byoc_req = ByocJobRequest( - capability=req.capability, - payload={"model_id": req.model_id, **req.params}, - timeout_seconds=req.timeout_seconds, - job_id=job_id, - ) - - # Sign the job request if signer is available — mirror submit_byoc_job. - # The orch's training handler runs setupOrchJob → verifyJobCreds (same - # code path as inference) and rejects with HTTP 400 "Could not verify - # job creds" if Livepeer-Job-Request / -Token are missing. The original - # unsigned path worked only against an older orch that lacked the - # /process/train/ route; the v2-with-training merge brings that route - # online and demands signed creds. - sender = "" - sig = "" - if signer_url: - try: - request_json = json.dumps(byoc_req.payload) - parameters_json = json.dumps(byoc_req.parameters) if byoc_req.parameters else "" - sign_resp = _sign_byoc_job( - signer_url=signer_url, - signer_headers=signer_headers, - job_id=job_id, - capability=req.capability, - request_json=request_json, - parameters_json=parameters_json, - timeout_seconds=req.timeout_seconds, - ) - sender = sign_resp.get("sender", "") - sig = sign_resp.get("signature", "") - _LOG.info("Training job %s: signed by sender=%s", job_id, - sender[:12] + "..." if sender else "none") - except Exception as e: - _LOG.warning("Training job %s: signing failed: %s", job_id, e) - - livepeer_hdr = _build_livepeer_header(byoc_req, job_id, sender=sender, sig=sig) - - # Build training body - body = json.dumps({ - "model_id": req.model_id, - "params": req.params, - **({"callback_url": req.callback_url} if req.callback_url else {}), - }).encode("utf-8") - - rejections: list[OrchestratorRejection] = [] - - for orch in orch_list: - orch_origin = _http_origin(orch) - url = f"{orch_origin}/process/train/{req.capability}" - - headers = { - "Content-Type": "application/json", - "Livepeer": livepeer_hdr, - "Livepeer-Capability": req.capability, - } - - # On-chain payment ticket — same flow as inference. Required for - # the orch's verifyJobCreds + per-second metering to succeed. - # On staging the capability price is 0 so deduction is a no-op, - # but the orch still validates the ticket structure. - if signer_url: - try: - payment_headers = _create_byoc_payment( - orch_origin=orch_origin, - capability=req.capability, - livepeer_hdr=livepeer_hdr, - signer_url=signer_url, - signer_headers=signer_headers, - timeout=http_timeout, - ) - headers.update(payment_headers) - _LOG.info("Training job %s: payment tickets created for %s", - job_id, orch_origin) - except Exception as e: - _LOG.warning("Training job %s: payment creation failed for %s: %s", - job_id, orch_origin, e) - rejections.append(OrchestratorRejection( - url=orch_origin, reason=f"payment failed: {e}", - )) - continue - - http_req = Request(url, data=body, headers=headers, method="POST") - _LOG.info("Training job %s: trying orchestrator %s", job_id, orch_origin) - - try: - with urlopen(http_req, timeout=http_timeout, context=_ssl_ctx) as resp: - raw_body = resp.read() - data = json.loads(raw_body.decode("utf-8")) - - return ByocTrainingResponse( - job_id=data.get("job_id", job_id), - status=data.get("status", "submitted"), - orchestrator_url=orch_origin, - status_url=data.get("status_url"), - data=data, - ) - - except HTTPError as e: - err_body = "" - try: - err_body = e.read().decode("utf-8", errors="replace")[:500] - except Exception: - pass - reason = f"HTTP {e.code}: {err_body}" - _LOG.warning("Training job %s: orchestrator %s rejected: %s", job_id, orch_origin, reason) - - if 400 <= e.code < 500 and e.code not in (408, 429): - raise LivepeerGatewayError( - f"Training job rejected by orchestrator {orch_origin}: {reason}" - ) from e - - rejections.append(OrchestratorRejection(url=orch_origin, reason=reason)) - - except (URLError, ConnectionRefusedError, TimeoutError, OSError) as e: - reason = f"{type(e).__name__}: {e}" - _LOG.warning("Training job %s: orchestrator %s unreachable: %s", job_id, orch_origin, reason) - rejections.append(OrchestratorRejection(url=orch_origin, reason=reason)) - - reasons = "; ".join(r.reason for r in rejections) if rejections else "no orchestrators configured" - raise NoOrchestratorAvailableError( - f"No orchestrator available for capability '{req.capability}': {reasons}", - rejections=rejections, - ) - - -def refresh_training_payment( - job_id: str, - orch_url: str, - capability: str, - *, - signer_url: str, - signer_headers: Optional[dict[str, str]] = None, - timeout: float = 30.0, - max_attempts: int = 3, -) -> dict[str, str]: - """ - Top up the orch's deposit ledger for an in-flight async training job. - - Called by the SDK's status-poll loop when the orch-reported balance - approaches zero (refresh-on-watermark per design §3.A). Generates a - fresh ticket batch from the same wallet that signed the submit, then - POSTs it to the orch at /process/job/{job_id}/refresh-payment. - - Invariants per §10.1: - - I5 (no double-charge on retry): orch idempotency key is - (job_id, ticket_nonce). The signer's /generate-live-payment - includes nonce in the payment payload; orch deduplicates. - - I6 (sender attribution): the refresh ticket is signed by the SAME - wallet as the submit (signer_url resolves bearer → wallet - deterministically; bearer is the same for the same SDK session). - - Args: - job_id: training job_id assigned by orch on submit. - orch_url: orchestrator URL accepting the job. - capability: capability name (needed by signer to pick correct - ticket_params). - signer_url: remote signer with /generate-live-payment. - signer_headers: pass-through headers (notably Authorization - Bearer of the user). - timeout: per-request timeout in seconds. - max_attempts: retry count for signer/orch transient errors. - Default 3 with simple linear backoff (no exponential — keeps - refresh latency bounded under SDK's watermark budget). - - Returns: - dict with at least {"credited_wei", "new_balance_wei"} fields - returned by the orch. Subset is reported to the SDK caller. - - Raises: - LivepeerGatewayError on permanent failure after max_attempts. - """ - # Mint the ticket ONCE, outside the retry loop. Re-minting on each - # retry would burn N distinct nonces for a single refresh attempt, - # and if the orch already credited the first ticket but the response - # was lost on the network, the second mint would double-credit. - # Reviewer note (I1): per-job idempotency at the PM layer keys on - # ticket nonce; identical headers credit once, distinct nonces credit - # separately. The fix is to never produce distinct nonces for a - # single logical refresh. - try: - payment_headers = _create_byoc_payment( - orch_origin=_http_origin(orch_url), - capability=capability, - livepeer_hdr="", # not used by refresh path - signer_url=signer_url, - signer_headers=signer_headers, - timeout=timeout, - ) - except LivepeerGatewayError as e: - # _create_byoc_payment raises on signer bug; surface as a refresh - # error rather than retrying (the signer state is what's wrong, - # not a network blip). - raise LivepeerGatewayError( - f"Training refresh {job_id}: payment generation failed: {e}" - ) from e - - if not payment_headers.get("Livepeer-Payment"): - # The orch's `ticket_params.face_value` is zero — the only way - # _create_byoc_payment returns an empty result without raising - # (it raises on signer-empty per the C1 fix). This means refresh - # is a no-op for this cap on this orch. - _LOG.info("Training refresh %s: orch face_value=0, noop", job_id) - return {"credited_wei": "0", "new_balance_wei": "n/a", "noop": "true"} - - # Retry only the orch POST. The payment headers above are pinned to - # one nonce; identical headers on every attempt → idempotent credit. - url = f"{_http_origin(orch_url)}/process/job/{job_id}/refresh-payment" - headers = { - "Content-Type": "application/json", - **payment_headers, # Livepeer-Payment + Livepeer-Segment - } - - last_err: Optional[Exception] = None - for attempt in range(1, max_attempts + 1): - # Empty JSON body — refresh carries everything in headers - http_req = Request(url, data=b"{}", headers=headers, method="POST") - try: - with urlopen(http_req, timeout=timeout, context=_ssl_ctx) as resp: - body = resp.read().decode("utf-8") - if resp.status not in (200, 202): - raise LivepeerGatewayError( - f"Refresh rejected: HTTP {resp.status}: {body[:200]}" - ) - try: - return json.loads(body) or {"credited_wei": "unknown"} - except json.JSONDecodeError: - # Older orch may return empty body; treat as success. - return {"credited_wei": "unknown", "raw": body[:200]} - - except (HTTPError, URLError, OSError) as e: - last_err = e - # HTTP 4xx (other than 408/429) are not transient — fail fast - if isinstance(e, HTTPError) and e.code not in (408, 429, 502, 503, 504): - err_body = "" - try: - err_body = e.read().decode("utf-8", errors="replace")[:200] - except Exception: - pass - raise LivepeerGatewayError( - f"Training refresh permanent failure for {job_id}: " - f"HTTP {e.code}: {err_body}" - ) from e - - _LOG.warning( - "Training refresh %s attempt %d/%d failed (%s); retrying", - job_id, attempt, max_attempts, type(e).__name__, - ) - if attempt < max_attempts: - import time - time.sleep(0.5 * attempt) # linear backoff: 0.5s, 1.0s - - raise LivepeerGatewayError( - f"Training refresh exhausted {max_attempts} attempts for {job_id}: {last_err}" - ) - - -def get_training_status( - job_id: str, - orch_url: str, - *, - timeout: float = 10.0, -) -> ByocTrainingStatus: - """ - Poll training job status from the orchestrator. - - Args: - job_id: The training job ID returned by submit_training_job. - orch_url: The orchestrator URL that accepted the job. - timeout: HTTP request timeout. - - Returns: - ByocTrainingStatus with current status, progress, and result. - """ - orch_origin = _http_origin(orch_url) - url = f"{orch_origin}/process/job/{job_id}" - http_req = Request(url, headers={"Accept": "application/json"}) - - try: - with urlopen(http_req, timeout=timeout, context=_ssl_ctx) as resp: - data = json.loads(resp.read().decode("utf-8")) - return ByocTrainingStatus( - job_id=data.get("job_id", job_id), - status=data.get("status", "unknown"), - progress=data.get("progress", 0), - result=data.get("result"), - error=data.get("error"), - model_id=data.get("model_id"), - created_at=data.get("created_at"), - updated_at=data.get("updated_at"), - cost=data.get("cost"), - balance=data.get("balance"), - ) - except HTTPError as e: - if e.code == 404: - raise LivepeerGatewayError(f"Training job {job_id} not found") from e - raise LivepeerGatewayError(f"Status check failed: HTTP {e.code}") from e - except Exception as e: - raise LivepeerGatewayError(f"Status check failed: {e}") from e - - -def wait_for_training( - job_id: str, - orch_url: str, - *, - poll_interval: float = 5.0, - timeout: float = 28800.0, -) -> ByocTrainingStatus: - """ - Poll until a training job completes. - - Args: - job_id: The training job ID. - orch_url: The orchestrator URL. - poll_interval: Seconds between polls. - timeout: Maximum wait time in seconds. - - Returns: - Final ByocTrainingStatus. - """ - import time - - elapsed = 0.0 - while elapsed < timeout: - status = get_training_status(job_id, orch_url) - if status.is_done: - return status - _LOG.info("Training job %s: status=%s progress=%d%% elapsed=%.0fs", - job_id, status.status, status.progress, elapsed) - time.sleep(poll_interval) - elapsed += poll_interval - - return get_training_status(job_id, orch_url) - - -# --------------------------------------------------------------------------- -# Internal helpers -# --------------------------------------------------------------------------- - -def _resolve_orchestrators( - *, - orch_url: Optional[Sequence[str] | str] = None, - discovery_url: Optional[str] = None, - signer_url: Optional[str] = None, - signer_headers: Optional[dict[str, str]] = None, - discovery_headers: Optional[dict[str, str]] = None, -) -> list[str]: - """Resolve orchestrator list from various sources.""" - # Direct orchestrator URL(s) - if orch_url is not None: - if isinstance(orch_url, str): - urls = [u.strip() for u in orch_url.split(",") if u.strip()] - else: - urls = [u.strip() for u in orch_url if isinstance(u, str) and u.strip()] - if urls: - return urls - - # Use discovery - if discovery_url or signer_url: - return discover_orchestrators( - discovery_url=discovery_url, - signer_url=signer_url, - signer_headers=signer_headers, - discovery_headers=discovery_headers, - ) - - raise LivepeerGatewayError( - "submit_byoc_job requires orch_url, discovery_url, or signer_url" - ) diff --git a/tests/test_byoc_refresh.py b/tests/test_byoc_refresh.py deleted file mode 100644 index f086080..0000000 --- a/tests/test_byoc_refresh.py +++ /dev/null @@ -1,305 +0,0 @@ -""" -Unit tests for refresh_training_payment helper (PR-2). - -Per design doc §11.1: -- P3: refresh_training_payment uses the same signer key as submit - (signer wallet doesn't change mid-job) → tested via header inspection -- Plus: idempotency invariant (I5 — duplicate refresh credits only once) -- Plus: 3-attempt retry on transient errors -- Plus: fail-fast on permanent 4xx -""" -from __future__ import annotations - -import json -from contextlib import contextmanager -from unittest.mock import MagicMock, patch -from urllib.error import HTTPError -from urllib.request import Request - -import pytest - -from livepeer_gateway.byoc import refresh_training_payment -from livepeer_gateway.errors import LivepeerGatewayError - - -def _stub_orch_info(): - info = MagicMock() - tp = MagicMock() - tp.face_value = b"\x01\x00" # non-zero - info.ticket_params = tp - info.HasField = lambda field: field == "ticket_params" - info.SerializeToString = lambda: b"stub-orch-info-protobuf" - return info - - -def _stub_orch_info_zero_price(): - """ticket_params with face_value=0 → signer says 'no payment needed'.""" - info = MagicMock() - tp = MagicMock() - tp.face_value = b"\x00" - info.ticket_params = tp - info.HasField = lambda field: field == "ticket_params" - info.SerializeToString = lambda: b"stub" - return info - - -@contextmanager -def _mock_http(*, signer_responses, orch_response_status=200, - orch_response_body=b'{"credited_wei":"1000","new_balance_wei":"2500"}', - orch_info=None): - captured: list[Request] = [] - - class _MockResponse: - def __init__(self, body: bytes, status: int = 200): - self._body = body - self.status = status - self.headers = {} - def read(self): return self._body - def __enter__(self): return self - def __exit__(self, *a): return False - - signer_iter = iter(signer_responses) - - def _fake_urlopen(req, *args, **kwargs): - captured.append(req) - url = req.full_url if hasattr(req, "full_url") else req.get_full_url() - if "signer" in url or "/generate-live-payment" in url: - try: payload = next(signer_iter) - except StopIteration: payload = {"payment": "MORE_TICKETS", "segCreds": "MORE_SEG"} - return _MockResponse(json.dumps(payload).encode()) - return _MockResponse(orch_response_body, orch_response_status) - - info = orch_info or _stub_orch_info() - with patch("livepeer_gateway.byoc.urlopen", side_effect=_fake_urlopen), \ - patch("livepeer_gateway.orch_info.get_orch_info", - side_effect=lambda *a, **k: info): - yield captured - - -# --------------------------------------------------------------------------- -# Happy path -# --------------------------------------------------------------------------- - - -def test_refresh_happy_path_credits_orch(): - """Refresh POSTs to /process/job//refresh-payment with payment headers.""" - with _mock_http( - signer_responses=[{"payment": "FRESH_TICKETS", "segCreds": "FRESH_SEG"}], - ) as reqs: - result = refresh_training_payment( - job_id="train-abc", - orch_url="https://orch.test:8935", - capability="flux-lora-training", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - ) - - refresh_req = [r for r in reqs if "refresh-payment" in r.full_url][0] - headers = {k.lower(): v for k, v in refresh_req.header_items()} - - assert "/process/job/train-abc/refresh-payment" in refresh_req.full_url - assert headers["livepeer-payment"] == "FRESH_TICKETS" - assert headers["livepeer-segment"] == "FRESH_SEG" - assert refresh_req.method == "POST" - assert result["credited_wei"] == "1000" - - -# --------------------------------------------------------------------------- -# P3 — refresh uses same signer wallet as submit -# --------------------------------------------------------------------------- - - -def test_p3_refresh_uses_same_signer_headers_as_submit(): - """ - Bearer header forwarded to signer is the SAME as caller passed in. - This ensures the signer resolves to the same wallet on submit + refresh - (per Invariant I6 — sender attribution). - """ - bearer = "Bearer sk_user_pulsex1" - with _mock_http( - signer_responses=[{"payment": "F", "segCreds": "S"}], - ) as reqs: - refresh_training_payment( - job_id="train-abc", - orch_url="https://orch.test:8935", - capability="flux-lora-training", - signer_url="https://signer.test", - signer_headers={"Authorization": bearer}, - ) - - signer_reqs = [r for r in reqs if "generate-live-payment" in r.full_url] - assert len(signer_reqs) == 1 - forwarded = {k.lower(): v for k, v in signer_reqs[0].header_items()} - assert forwarded.get("authorization") == bearer, ( - f"signer didn't receive caller's bearer; got {forwarded.get('authorization')!r}" - ) - - -# --------------------------------------------------------------------------- -# Zero-price case — refresh is no-op -# --------------------------------------------------------------------------- - - -def test_refresh_zero_price_is_noop(): - """If signer says face_value=0, the orch refresh POST is skipped (no-op).""" - with _mock_http( - signer_responses=[{"payment": "", "segCreds": ""}], # ignored - orch_info=_stub_orch_info_zero_price(), - ) as reqs: - result = refresh_training_payment( - job_id="train-zero", - orch_url="https://orch.test:8935", - capability="flux-lora-training", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - ) - - refresh_calls = [r for r in reqs if "refresh-payment" in r.full_url] - assert refresh_calls == [], "refresh fired against orch despite zero-price" - assert result.get("noop") == "true" - assert result["credited_wei"] == "0" - - -# --------------------------------------------------------------------------- -# Retry on transient orch error -# --------------------------------------------------------------------------- - - -def test_refresh_retries_on_503(): - """Transient 503 from orch → retry (up to max_attempts), eventually succeed. - - Reviewer I2 strengthening: also verify that signer is called ONCE - (not once per retry) and that all retry attempts send the SAME - Livepeer-Payment header value (no re-minting → no nonce drift). - """ - call_count = {"signer": 0, "orch": 0} - orch_payment_headers: list[str] = [] - - class _MockResponse: - def __init__(self, body, status=200): - self._body, self.status = body, status - self.headers = {} - def read(self): return self._body - def __enter__(self): return self - def __exit__(self, *a): return False - - def _fake_urlopen(req, *args, **kwargs): - url = req.full_url if hasattr(req, "full_url") else req.get_full_url() - if "generate-live-payment" in url: - call_count["signer"] += 1 - return _MockResponse(json.dumps({"payment": "PINNED_T", "segCreds": "PINNED_S"}).encode()) - # orch refresh — capture the Livepeer-Payment header so we can - # assert all retries used the same ticket batch - hdrs = {k.lower(): v for k, v in req.header_items()} - orch_payment_headers.append(hdrs.get("livepeer-payment", "")) - call_count["orch"] += 1 - if call_count["orch"] < 3: - raise HTTPError(url, 503, "service unavailable", {}, None) - return _MockResponse(b'{"credited_wei":"500"}', 200) - - with patch("livepeer_gateway.byoc.urlopen", side_effect=_fake_urlopen), \ - patch("livepeer_gateway.orch_info.get_orch_info", - side_effect=lambda *a, **k: _stub_orch_info()): - result = refresh_training_payment( - job_id="train-retry", - orch_url="https://orch.test:8935", - capability="flux-lora-training", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - max_attempts=3, - ) - - assert call_count["orch"] == 3, f"expected 3 orch attempts, got {call_count['orch']}" - assert call_count["signer"] == 1, ( - f"signer was called {call_count['signer']} times; should mint ONCE outside retry " - "loop to avoid nonce drift (Invariant I5)" - ) - assert len(set(orch_payment_headers)) == 1, ( - f"orch attempts sent different Livepeer-Payment headers: {orch_payment_headers}; " - "all retries must reuse the same ticket batch" - ) - assert orch_payment_headers[0] == "PINNED_T", ( - f"expected pinned ticket value 'PINNED_T', got {orch_payment_headers[0]!r}" - ) - assert result["credited_wei"] == "500" - - -# --------------------------------------------------------------------------- -# Fail-fast on permanent 4xx (e.g., bad job_id, expired token) -# --------------------------------------------------------------------------- - - -def test_refresh_fails_fast_on_permanent_4xx(): - """403/404 from orch → no retry, raise immediately.""" - call_count = {"orch": 0} - - class _MockResponse: - def __init__(self, body, status=200): - self._body, self.status = body, status - self.headers = {} - def read(self): return self._body - def __enter__(self): return self - def __exit__(self, *a): return False - - def _fake_urlopen(req, *args, **kwargs): - url = req.full_url if hasattr(req, "full_url") else req.get_full_url() - if "generate-live-payment" in url: - return _MockResponse(json.dumps({"payment": "T", "segCreds": "S"}).encode()) - call_count["orch"] += 1 - err = HTTPError(url, 403, "sender mismatch", {}, None) - # Make HTTPError.read() return useful body - err.read = lambda: b"sender mismatch with original submit" - raise err - - with patch("livepeer_gateway.byoc.urlopen", side_effect=_fake_urlopen), \ - patch("livepeer_gateway.orch_info.get_orch_info", - side_effect=lambda *a, **k: _stub_orch_info()): - with pytest.raises(LivepeerGatewayError) as excinfo: - refresh_training_payment( - job_id="train-403", - orch_url="https://orch.test:8935", - capability="flux-lora-training", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - max_attempts=3, - ) - - assert "permanent failure" in str(excinfo.value).lower() - assert call_count["orch"] == 1, "fail-fast on 403 should not retry" - - -# --------------------------------------------------------------------------- -# Exhaustion — all 3 attempts fail -# --------------------------------------------------------------------------- - - -def test_refresh_exhausts_retries(): - """All 3 attempts fail with 503 → raise LivepeerGatewayError.""" - class _MockResponse: - def __init__(self, body, status=200): - self._body, self.status = body, status - self.headers = {} - def read(self): return self._body - def __enter__(self): return self - def __exit__(self, *a): return False - - def _fake_urlopen(req, *args, **kwargs): - url = req.full_url if hasattr(req, "full_url") else req.get_full_url() - if "generate-live-payment" in url: - return _MockResponse(json.dumps({"payment": "T", "segCreds": "S"}).encode()) - raise HTTPError(url, 503, "always down", {}, None) - - with patch("livepeer_gateway.byoc.urlopen", side_effect=_fake_urlopen), \ - patch("livepeer_gateway.orch_info.get_orch_info", - side_effect=lambda *a, **k: _stub_orch_info()): - with pytest.raises(LivepeerGatewayError) as excinfo: - refresh_training_payment( - job_id="train-doomed", - orch_url="https://orch.test:8935", - capability="flux-lora-training", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - max_attempts=3, - ) - - assert "exhausted" in str(excinfo.value).lower() diff --git a/tests/test_byoc_training.py b/tests/test_byoc_training.py deleted file mode 100644 index 52d66ff..0000000 --- a/tests/test_byoc_training.py +++ /dev/null @@ -1,220 +0,0 @@ -""" -Unit tests for submit_training_job sign + payment flow (PR-1). - -Per design doc §11.1, P1-P3: -- P1: submit_training_job emits the same 4 headers as submit_byoc_job -- P2: submit_training_job with signer_url=None proceeds with empty creds -- P3: refresh_training_payment uses same signer key (covered in test_byoc_refresh.py) - -These tests mock urllib.request.urlopen to capture the outgoing Request and -inspect headers/body without touching network. -""" -from __future__ import annotations - -import json -from contextlib import contextmanager -from unittest.mock import MagicMock, patch -from urllib.request import Request - -from livepeer_gateway.byoc import ( - ByocJobRequest, - ByocTrainingRequest, - submit_byoc_job, - submit_training_job, -) - - -# --------------------------------------------------------------------------- -# Test fixtures -# --------------------------------------------------------------------------- - - -def _stub_orch_info(): - """OrchestratorInfo-like object with non-zero ticket params.""" - info = MagicMock() - tp = MagicMock() - tp.face_value = b"\x01\x00" # non-zero → payment generation proceeds - info.ticket_params = tp - info.HasField = lambda field: field == "ticket_params" - info.SerializeToString = lambda: b"stub-orch-info-protobuf" - return info - - -@contextmanager -def _mock_http(*, signer_responses, orch_response_status=200, - orch_response_body=b'{"status":"submitted","job_id":"orch-123","status_url":"/process/job/orch-123"}'): - """ - Mock urllib.request.urlopen used inside byoc.py. signer_responses are - consumed in order on signer-host calls. Orch call returns the response_*. - - Captures every Request object on the yielded list for assertion. - """ - captured_requests: list[Request] = [] - - class _MockResponse: - def __init__(self, body: bytes, status: int = 200): - self._body = body - self.status = status - self.headers = {} - - def read(self): - return self._body - - def __enter__(self): - return self - - def __exit__(self, *args): - return False - - signer_iter = iter(signer_responses) - - def _fake_urlopen(req, *args, **kwargs): - captured_requests.append(req) - url = req.full_url if hasattr(req, "full_url") else req.get_full_url() - - if "signer" in url or "/sign-byoc-job" in url or "/generate-live-payment" in url: - try: - payload = next(signer_iter) - except StopIteration: - payload = {"sender": "0xMOCKSENDER", "signature": "0xMOCKSIG"} - return _MockResponse(json.dumps(payload).encode()) - return _MockResponse(orch_response_body, orch_response_status) - - with patch("livepeer_gateway.byoc.urlopen", side_effect=_fake_urlopen), \ - patch("livepeer_gateway.byoc._resolve_orchestrators", - return_value=["https://orch.test:8935"]), \ - patch("livepeer_gateway.orch_info.get_orch_info", - side_effect=lambda *a, **k: _stub_orch_info()): - yield captured_requests - - -# --------------------------------------------------------------------------- -# P1 — header parity between training and inference paths -# --------------------------------------------------------------------------- - - -def test_p1_training_job_emits_same_headers_as_inference(): - """ - Submit a training job AND an inference job with the same signer config. - Both must produce the same 4 critical headers on the orch request. - """ - # 1. inference path - with _mock_http(signer_responses=[ - {"sender": "0xWALLET1", "signature": "0xSIGABC"}, - {"payment": "TICKETS_B64", "segCreds": "SEG_B64"}, - ]) as inf_reqs: - submit_byoc_job( - req=ByocJobRequest(capability="flux-dev", payload={"prompt": "x"}, job_id="job-inf-1"), - orch_url="https://orch.test:8935", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - ) - - inf_orch_req = [r for r in inf_reqs if "process/request" in r.full_url][0] - inf_headers = {k.lower(): v for k, v in inf_orch_req.header_items()} - - # 2. training path - with _mock_http( - signer_responses=[ - {"sender": "0xWALLET1", "signature": "0xSIGABC"}, - {"payment": "TICKETS_B64", "segCreds": "SEG_B64"}, - ], - orch_response_status=202, - ) as tr_reqs: - submit_training_job( - req=ByocTrainingRequest( - capability="flux-lora-training", - model_id="flux-dev", - params={"images_data_url": "https://x/zip", "trigger_word": "TOK", "steps": 10}, - ), - orch_url="https://orch.test:8935", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - ) - - tr_orch_req = [r for r in tr_reqs if "process/train" in r.full_url][0] - tr_headers = {k.lower(): v for k, v in tr_orch_req.header_items()} - - required = {"livepeer", "livepeer-capability", "livepeer-payment", "livepeer-segment"} - missing_inf = required - inf_headers.keys() - missing_tr = required - tr_headers.keys() - - assert not missing_inf, f"inference path missing headers: {missing_inf}" - assert not missing_tr, f"training path missing headers: {missing_tr}" - - assert inf_headers["livepeer-payment"], "inference Livepeer-Payment was empty" - assert tr_headers["livepeer-payment"], "training Livepeer-Payment was empty" - assert inf_headers["livepeer-segment"], "inference Livepeer-Segment was empty" - assert tr_headers["livepeer-segment"], "training Livepeer-Segment was empty" - - -# --------------------------------------------------------------------------- -# P2 — offchain mode (no signer_url) does NOT emit payment headers -# --------------------------------------------------------------------------- - - -def test_p2_training_no_signer_proceeds_unsigned(): - """ - With signer_url=None, training submit must skip signing AND payment, - and still POST to orch. Mirrors submit_byoc_job's behavior. - """ - with _mock_http( - signer_responses=[], - orch_response_status=202, - ) as reqs: - submit_training_job( - req=ByocTrainingRequest( - capability="flux-lora-training", - model_id="flux-dev", - params={"images_data_url": "https://x/zip", "trigger_word": "TOK", "steps": 10}, - ), - orch_url="https://orch.test:8935", - signer_url=None, - ) - - signer_calls = [r for r in reqs if "signer" in r.full_url or "sign-byoc-job" in r.full_url] - assert signer_calls == [], ( - f"unexpected signer calls in offchain mode: {[r.full_url for r in signer_calls]}" - ) - - orch_calls = [r for r in reqs if "process/train" in r.full_url] - assert len(orch_calls) == 1, "expected exactly one /process/train POST" - - headers = {k.lower(): v for k, v in orch_calls[0].header_items()} - assert "livepeer" in headers - assert headers.get("livepeer-capability") == "flux-lora-training" - assert "livepeer-payment" not in headers, "Livepeer-Payment leaked into offchain submit" - assert "livepeer-segment" not in headers, "Livepeer-Segment leaked into offchain submit" - - -# --------------------------------------------------------------------------- -# Bonus — body shape -# --------------------------------------------------------------------------- - - -def test_training_body_includes_model_id_and_params(): - """submit_training_job sends model_id + params at top level of body.""" - with _mock_http( - signer_responses=[ - {"sender": "0xW", "signature": "0xS"}, - {"payment": "T", "segCreds": "S"}, - ], - orch_response_status=202, - ) as reqs: - submit_training_job( - req=ByocTrainingRequest( - capability="flux-lora-training", - model_id="flux-dev", - params={"images_data_url": "https://x/zip", "trigger_word": "PULSEX1", "steps": 1000}, - ), - orch_url="https://orch.test:8935", - signer_url="https://signer.test", - signer_headers={"Authorization": "Bearer sk_test"}, - ) - - orch_req = [r for r in reqs if "process/train" in r.full_url][0] - body = json.loads(orch_req.data.decode()) - assert body["model_id"] == "flux-dev" - assert body["params"]["images_data_url"] == "https://x/zip" - assert body["params"]["trigger_word"] == "PULSEX1" - assert body["params"]["steps"] == 1000