feat(signer): BYOC per-capability pricing + real capability/model usage labels

byoc/job_orchestrator.go

@@ -566,18 +566,26 @@ func (bso *BYOCOrchestratorServer) verifyJobCreds(ctx context.Context, jobCreds
 		return nil, errSegSig
 	}
 
-	if !bso.orch.VerifySig(ethcommon.HexToAddress(jobData.Sender), jobData.Request+jobData.Parameters, sigByte) {
-		clog.Errorf(ctx, "Sig check failed sender=%v", jobData.Sender)
-		return nil, errSegSig
-	}
-
-	if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil {
-		return nil, errZeroCapacity
+	sender := ethcommon.HexToAddress(jobData.Sender)
+
+	// Verify V1 structured binary format (matches DMZ /sign-byoc-job signing).
+	v1Payload := FlattenBYOCJob(&BYOCJobSigningInput{
+		ID:             jobData.ID,
+		Capability:     jobData.Capability,
+		Request:        jobData.Request,
+		Parameters:     jobData.Parameters,
+		TimeoutSeconds: jobData.Timeout,
+	})
+	if bso.orch.VerifySig(sender, string(v1Payload), sigByte) {
+		if reserveCapacity && bso.orch.ReserveExternalCapabilityCapacity(jobData.Capability) != nil {
+			return nil, errZeroCapacity
+		}
+		jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability)
+		return jobData, nil
 	}
 
-	jobData.CapabilityUrl = bso.orch.GetUrlForCapability(jobData.Capability)
-
-	return jobData, nil
+	clog.Errorf(ctx, "Sig check failed sender=%v", jobData.Sender)
+	return nil, errSegSig
 }
 
 func (bso *BYOCOrchestratorServer) verifyTokenCreds(ctx context.Context, tokenCreds string) (*JobSender, error) {

cmd/livepeer/starter/flags.go

@@ -149,6 +149,7 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig {
 	cfg.RemoteSignerWebhookHeaders = fs.String("remoteSignerWebhookHeaders", *cfg.RemoteSignerWebhookHeaders, "Map of headers to use for remote signer webhook requests. e.g. 'header:val,header2:val2'")
 	cfg.RemoteSignerAllowNoAuth = fs.Bool("remoteSignerAllowNoAuth", *cfg.RemoteSignerAllowNoAuth, "Allow an unauthenticated remote signer on a public -httpAddr (no webhook). UNSAFE: signs payments from this node's deposit for any reachable caller; restrict access externally (proxy/private network).")
 	cfg.RemoteDiscovery = fs.Bool("remoteDiscovery", *cfg.RemoteDiscovery, "Enable orchestrator discovery on remote signers")
+	cfg.ByocPerCapPricing = fs.Bool("byocPerCapPricing", *cfg.ByocPerCapPricing, "Remote signer: resolve BYOC live-payment fee from the orchestrator's per-capability CapabilitiesPrices (keyed on the request capability) instead of the flat base price. Default OFF (byte-identical to base-price behavior).")
 
 	// Gateway metrics
 	cfg.KafkaBootstrapServers = fs.String("kafkaBootstrapServers", *cfg.KafkaBootstrapServers, "URL of Kafka Bootstrap Servers")

cmd/livepeer/starter/starter.go

@@ -180,6 +180,7 @@ type LivepeerConfig struct {
 	RemoteSignerWebhookHeaders *string
 	RemoteSignerAllowNoAuth    *bool
 	RemoteDiscovery            *bool
+	ByocPerCapPricing          *bool
 	AIRunnerImage              *string
 	AIRunnerImageOverrides     *string
 	AIVerboseLogs              *bool
@@ -326,6 +327,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 	defaultRemoteSignerWebhookHeaders := ""
 	defaultRemoteSignerAllowNoAuth := false
 	defaultRemoteDiscovery := false
+	defaultByocPerCapPricing := false
 
 	// Gateway logs
 	defaultKafkaBootstrapServers := ""
@@ -458,6 +460,7 @@ func DefaultLivepeerConfig() LivepeerConfig {
 		RemoteSignerWebhookHeaders: &defaultRemoteSignerWebhookHeaders,
 		RemoteSignerAllowNoAuth:    &defaultRemoteSignerAllowNoAuth,
 		RemoteDiscovery:            &defaultRemoteDiscovery,
+		ByocPerCapPricing:          &defaultByocPerCapPricing,
 
 		// Gateway logs
 		KafkaBootstrapServers: &defaultKafkaBootstrapServers,
@@ -1881,6 +1884,9 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	if cfg.RemoteDiscovery != nil {
 		n.RemoteDiscovery = *cfg.RemoteDiscovery
 	}
+	if cfg.ByocPerCapPricing != nil {
+		n.ByocPerCapPricing = *cfg.ByocPerCapPricing
+	}
 	if cfg.LiveAIHeartbeatHeaders != nil {
 		n.LiveAIHeartbeatHeaders = parseHeaderMap(*cfg.LiveAIHeartbeatHeaders)
 	}

core/livepeernode.go

@@ -158,6 +158,7 @@ type LivepeerNode struct {
 	RemoteEthAddr              ethcommon.Address // eth address of the remote signer
 	InfoSig                    []byte            // sig over eth address for the OrchestratorInfo request
 	RemoteDiscovery            bool              // expose remote discovery endpoint when enabled
+	ByocPerCapPricing          bool              // resolve BYOC fee from per-capability CapabilitiesPrices instead of base price (remote signer; default OFF)
 
 	// Thread safety for config fields
 	mu                  sync.RWMutex

server/remote_signer.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"math/big"
 	"net/http"
 	"net/url"
@@ -33,6 +34,7 @@ const HTTPStatusPriceExceeded = 481
 const HTTPStatusNoTickets = 482
 const RefreshSessionOrchestratorURLHeader = "Livepeer-Orchestrator-URL"
 const RemoteType_LiveVideoToVideo = "lv2v"
+const RemoteType_BYOC = "byoc"
 const PipelineLiveVideoToVideo = "live-video-to-video"
 const remoteSignerAuthIDHeader = "Signer-Auth-Id"
 
@@ -242,11 +244,25 @@ type RemotePaymentRequest struct {
 	// Number of pixels to generate a ticket for. Required if `type` is not set.
 	InPixels int64 `json:"inPixels"`
 
-	// Job type to automatically calculate payments. Valid values: `lv2v`. Optional.
+	// Job type to automatically calculate payments. Valid values: `lv2v`, `byoc`. Optional.
 	Type string `json:"type"`
 
 	// Capabilities to include in the ticket. Optional; may be set for the lv2v job type.
 	Capabilities []byte `json:"capabilities"`
+
+	// Capability is the real BYOC capability name (e.g. "nano-banana"). When
+	// set, it overrides the metered `pipeline` label for the emitted
+	// create_signed_ticket usage event, decoupling usage attribution from
+	// `Type` (which stays "lv2v" for fee/pixel routing). Optional; empty
+	// preserves the previous behavior (pipeline falls back to the lv2v
+	// constant when Type=="lv2v").
+	Capability string `json:"capability,omitempty"`
+
+	// ModelID is the specific provider model from the request
+	// payload/parameters. When set, it overrides the metered `model_id`
+	// label. Optional; empty preserves the previous behavior (the
+	// capabilities-derived model id, which defaults to "unknown" downstream).
+	ModelID string `json:"model_id,omitempty"`
 }
 
 // Returned by the remote signer and includes a new payment plus updated state.
@@ -355,6 +371,120 @@ func (ls *LivepeerServer) authLivePayment(r *http.Request, state *RemotePaymentS
 	return *webhookResp.Status, &webhookResp, errors.New(webhookResp.Reason)
 }
 
+// maxUsageLabelLen bounds the length of gateway-supplied usage labels
+// (pipeline/model_id) before they are emitted to the metering pipeline.
+// req.Capability / req.ModelID come straight from the request body, so a buggy
+// or malicious caller could otherwise send very long / high-cardinality values
+// that inflate Kafka payload size and downstream label cardinality (the
+// pipeline label was previously a small fixed set). 128 runes is generous for
+// real capability/model identifiers.
+const maxUsageLabelLen = 128
+
+// sanitizeUsageLabel trims surrounding whitespace and caps the length (in
+// runes, to never emit invalid UTF-8) of a gateway-supplied metering label.
+func sanitizeUsageLabel(s string) string {
+	s = strings.TrimSpace(s)
+	if r := []rune(s); len(r) > maxUsageLabelLen {
+		return string(r[:maxUsageLabelLen])
+	}
+	return s
+}
+
+// resolveUsageLabels derives the metering `pipeline` and `model_id` labels for
+// the emitted create_signed_ticket usage event.
+//
+// The real BYOC capability/model supplied by the gateway (req.Capability /
+// req.ModelID) take precedence and override the labels whenever they are set,
+// regardless of `Type` — this decouples usage attribution from `Type`, which
+// still drives fee/pixel routing. The gateway-supplied values are sanitized
+// (trimmed + length-capped) to bound payload size and label cardinality.
+//
+// When those additive fields are empty, the labels fall back to the previous
+// behavior: for lv2v the pipeline is the live-video-to-video constant and the
+// model id is derived from the request capabilities; for any other request the
+// untouched label(s) remain empty (the collector then defaults them to
+// "unknown"). This makes non-BYOC (lv2v) callers with no override
+// byte-identical to before.
+func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pipeline, modelID string) {
+	if req.Type == RemoteType_LiveVideoToVideo {
+		pipeline = PipelineLiveVideoToVideo
+		if caps != nil {
+			modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo)
+		}
+	}
+	if req.Type == RemoteType_BYOC && caps != nil {
+		if m := caps.ModelIDForCapability(core.Capability_BYOC); m != "" {
+			pipeline = core.CapabilityToPipeline(core.Capability_BYOC)
+			modelID = m
+		}
+	}
+	if c := sanitizeUsageLabel(req.Capability); c != "" {
+		pipeline = c
+	}
+	if m := sanitizeUsageLabel(req.ModelID); m != "" {
+		modelID = m
+	}
+	return pipeline, modelID
+}
+
+// byocCapabilityName returns the BYOC capability constraint used for per-cap
+// pricing lookup. The gateway may supply it as req.Capability (lv2v path) or
+// encode it in the BYOC capabilities protobuf (type:"byoc" path).
+func byocCapabilityName(req *RemotePaymentRequest, caps *core.Capabilities) string {
+	if c := sanitizeUsageLabel(req.Capability); c != "" {
+		return c
+	}
+	if caps != nil {
+		return caps.ModelIDForCapability(core.Capability_BYOC)
+	}
+	return ""
+}
+
+func isByocBillingType(t string) bool {
+	return t == RemoteType_BYOC || t == RemoteType_LiveVideoToVideo
+}
+
+// resolveByocPrice resolves the per-capability BYOC price advertised in the
+// orchestrator's OrchestratorInfo.CapabilitiesPrices, keyed on the request
+// capability name (req.Capability). The orchestrator appends external BYOC
+// capability prices to CapabilitiesPrices as
+// {Capability: Capability_BYOC, Constraint: <capability name>} (see
+// core/orchestrator.go GetCapabilitiesPrices), so a match is a BYOC entry whose
+// Constraint equals req.Capability.
+//
+// Granularity is per-capability only: req.ModelID is intentionally NOT used for
+// price selection (model_id still flows through for metering attribution).
+//
+// Returns nil when there is no usable per-capability price (no capability set,
+// or no matching entry with a positive rate). A matching entry with a
+// non-positive rate is skipped rather than aborting the scan, so a later valid
+// duplicate entry for the same constraint (e.g. due to misconfiguration) is
+// still honored. Callers fall back to the base oInfo.PriceInfo when nil is
+// returned, preserving today's behavior for unconfigured or misconfigured
+// capabilities. This is a pure function (no flag, no IO) so it is hermetically
+// testable without the CGO/ffmpeg toolchain.
+func resolveByocPrice(req *RemotePaymentRequest, oInfo *net.OrchestratorInfo) *net.PriceInfo {
+	if req == nil || oInfo == nil || req.Capability == "" {
+		return nil
+	}
+	for _, p := range oInfo.CapabilitiesPrices {
+		if p == nil {
+			continue
+		}
+		if p.Capability != uint32(core.Capability_BYOC) || p.Constraint != req.Capability {
+			continue
+		}
+		// Only honor a valid positive rate. Skip invalid/zero entries and keep
+		// scanning so a later valid duplicate for the same constraint isn't
+		// shadowed; if none is found we fall back to base (never zeroing the fee).
+		if p.PricePerUnit <= 0 || p.PixelsPerUnit <= 0 {
+			continue
+		}
+		return &net.PriceInfo{PricePerUnit: p.PricePerUnit, PixelsPerUnit: p.PixelsPerUnit}
+	}
+	return nil
+}
+
 // GenerateLivePayment handles remote generation of a payment for live streams.
 func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Request) {
 	requestID := string(core.RandomManifestID())
@@ -389,7 +519,43 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
 		return
 	}
+	// BYOC per-capability pricing (flag-gated, default OFF). When enabled and the
+	// gateway supplied a real capability, resolve the per-capability price from
+	// the orchestrator's advertised CapabilitiesPrices and use it instead of the
+	// flat base price. The resolved price is written back to oInfo.PriceInfo so it
+	// is the single source for: state init, initialPrice, the max-price ceiling,
+	// the payment's ExpectedPrice (which the orchestrator uses to set its fixed
+	// per-session price), and the price-doubling guard in validatePrice (which
+	// reads sess.OrchestratorInfo.PriceInfo) — keeping all of them cap-vs-cap
+	// consistent. When the flag is OFF or no usable cap price matches, behavior is
+	// byte-identical to the base-price path.
 	priceInfo := oInfo.PriceInfo
+	useByocPricing := false
+	var parsedCaps *core.Capabilities
+	if len(req.Capabilities) > 0 {
+		var caps net.Capabilities
+		if err := proto.Unmarshal(req.Capabilities, &caps); err != nil {
+			clog.Errorf(ctx, "Failed to unmarshal capabilities err=%q", err)
+			respondJsonError(ctx, w, err, http.StatusBadRequest)
+			return
+		}
+		parsedCaps = core.CapabilitiesFromNetCapabilities(&caps)
+	}
+	capName := byocCapabilityName(&req, parsedCaps)
+	// BYOC per-cap pricing changes the billing basis (compute-seconds instead
+	// of lv2v pixels). Apply for lv2v or explicit byoc job types when a
+	// capability constraint is present.
+	if ls.LivepeerNode.ByocPerCapPricing && capName != "" && isByocBillingType(req.Type) {
+		priceReq := req
+		if priceReq.Capability == "" {
+			priceReq.Capability = capName
+		}
+		if capPrice := resolveByocPrice(&priceReq, &oInfo); capPrice != nil {
+			priceInfo = capPrice
+			oInfo.PriceInfo = capPrice
+			useByocPricing = true
+		}
+	}
 	if priceInfo == nil || priceInfo.PricePerUnit == 0 || priceInfo.PixelsPerUnit == 0 {
 		err := fmt.Errorf("missing or zero priceInfo")
 		respondJsonError(ctx, w, err, http.StatusBadRequest)
@@ -456,14 +622,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		// Embedded within genSegCreds, may be used by orch for payment accounting
 		ManifestID: core.ManifestID(manifestID),
 	}
-	if len(req.Capabilities) > 0 {
-		var caps net.Capabilities
-		if err := proto.Unmarshal(req.Capabilities, &caps); err != nil {
-			clog.Errorf(ctx, "Failed to unmarshal capabilities err=%q", err)
-			respondJsonError(ctx, w, err, http.StatusBadRequest)
-			return
-		}
-		streamParams.Capabilities = core.CapabilitiesFromNetCapabilities(&caps)
+	if parsedCaps != nil {
+		streamParams.Capabilities = parsedCaps
 	}
 
 	pmParams := pmTicketParams(oInfo.TicketParams)
@@ -528,7 +688,17 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		lastUpdate = now
 	}
 	billableSecs := now.Sub(lastUpdate).Seconds()
-	if req.Type == RemoteType_LiveVideoToVideo {
+	if useByocPricing || req.Type == RemoteType_BYOC {
+		// BYOC per-capability tariff is denominated per compute-second (wei/sec),
+		// so charge on seconds directly rather than synthesizing lv2v pixels. With
+		// pixels = ceil(billableSecs) and the resolved per-cap price kept as its
+		// reduced wei/sec rational, calculateFee yields capPriceWeiPerSec * seconds.
+		if billableSecs <= 0 {
+			// preload with 60 seconds, mirroring the lv2v first-call behavior
+			billableSecs = (60 * time.Second).Seconds()
+		}
+		pixels = int64(math.Ceil(billableSecs))
+	} else if req.Type == RemoteType_LiveVideoToVideo {
 		info := defaultSegInfo
 		if billableSecs <= 0 {
 			// preload with 60 seconds of data for LV2V