fix(signer): meter real BYOC capability + model_id for usage events

server/remote_signer.go

@@ -247,6 +247,20 @@ type RemotePaymentRequest struct {
 
 	// 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 +369,54 @@ 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
+		modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo)
+	}
+	if c := sanitizeUsageLabel(req.Capability); c != "" {
+		pipeline = c
+	}
+	if m := sanitizeUsageLabel(req.ModelID); m != "" {
+		modelID = m
+	}
+	return pipeline, modelID
+}
+
 // GenerateLivePayment handles remote generation of a payment for live streams.
 func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Request) {
 	requestID := string(core.RandomManifestID())
@@ -680,12 +742,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		if state.SequenceNumber == 0 {
 			sessionStatus = "new"
 		}
-		pipeline := ""
-		modelID := ""
-		if req.Type == RemoteType_LiveVideoToVideo {
-			pipeline = PipelineLiveVideoToVideo
-			modelID = streamParams.Capabilities.ModelIDForCapability(core.Capability_LiveVideoToVideo)
-		}
+		pipeline, modelID := resolveUsageLabels(&req, streamParams.Capabilities)
 		// NB: This could could drop events if tha Kafka queue is full!
 		monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{
 			"session_id":         state.StateID,

server/remote_signer_test.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"strings"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -1692,3 +1693,110 @@ func TestGetOrchInfoSig_SendsConfiguredHeaders(t *testing.T) {
 	require.Equal([]byte{0x12, 0x34}, []byte(resp.Address))
 	require.Equal([]byte{0xab, 0xcd}, []byte(resp.Signature))
 }
+
+// lv2vCapsWithModel builds a core.Capabilities advertising a single warm model
+// for the live-video-to-video capability (used to exercise the existing
+// capabilities-derived model_id fallback).
+func lv2vCapsWithModel(t *testing.T, modelID string) *core.Capabilities {
+	t.Helper()
+	c := core.NewCapabilities([]core.Capability{core.Capability_LiveVideoToVideo}, nil)
+	c.SetPerCapabilityConstraints(core.PerCapabilityConstraints{
+		core.Capability_LiveVideoToVideo: &core.CapabilityConstraints{
+			Models: core.ModelConstraints{modelID: &core.ModelConstraint{Warm: true}},
+		},
+	})
+	return c
+}
+
+// TestResolveUsageLabels asserts the additive usage-attribution contract:
+//   - when the gateway supplies the real BYOC capability/model, the metered
+//     pipeline + model_id reflect them (the root-cause fix);
+//   - when those fields are empty, behavior is unchanged (lv2v constant +
+//     capabilities-derived model id, or empty → "unknown" downstream).
+//
+// This is hermetic: it exercises the pure label-resolution helper without the
+// CGO ffmpeg toolchain, the remote-signer HTTP handler, or Kafka.
+func TestResolveUsageLabels(t *testing.T) {
+	tests := []struct {
+		name         string
+		req          RemotePaymentRequest
+		caps         *core.Capabilities
+		wantPipeline string
+		wantModelID  string
+	}{
+		{
+			name:         "lv2v unchanged: no new fields, no caps",
+			req:          RemotePaymentRequest{Type: RemoteType_LiveVideoToVideo},
+			caps:         nil,
+			wantPipeline: PipelineLiveVideoToVideo,
+			wantModelID:  "",
+		},
+		{
+			name:         "lv2v unchanged: model derived from capabilities",
+			req:          RemotePaymentRequest{Type: RemoteType_LiveVideoToVideo},
+			caps:         lv2vCapsWithModel(t, "streamdiffusion"),
+			wantPipeline: PipelineLiveVideoToVideo,
+			wantModelID:  "streamdiffusion",
+		},
+		{
+			name: "byoc: real capability + model override lv2v defaults",
+			req: RemotePaymentRequest{
+				Type:       RemoteType_LiveVideoToVideo,
+				Capability: "nano-banana",
+				ModelID:    "google/nano-banana",
+			},
+			caps:         lv2vCapsWithModel(t, "streamdiffusion"),
+			wantPipeline: "nano-banana",
+			wantModelID:  "google/nano-banana",
+		},
+		{
+			name: "byoc: capability set, empty model falls back to caps-derived",
+			req: RemotePaymentRequest{
+				Type:       RemoteType_LiveVideoToVideo,
+				Capability: "nano-banana",
+			},
+			caps:         lv2vCapsWithModel(t, "streamdiffusion"),
+			wantPipeline: "nano-banana",
+			wantModelID:  "streamdiffusion",
+		},
+		{
+			name:         "non-lv2v with no fields: both empty (collector defaults to unknown)",
+			req:          RemotePaymentRequest{},
+			caps:         nil,
+			wantPipeline: "",
+			wantModelID:  "",
+		},
+		{
+			name: "byoc: whitespace-only override is ignored, lv2v defaults kept",
+			req: RemotePaymentRequest{
+				Type:       RemoteType_LiveVideoToVideo,
+				Capability: "   ",
+				ModelID:    "\t\n",
+			},
+			caps:         lv2vCapsWithModel(t, "streamdiffusion"),
+			wantPipeline: PipelineLiveVideoToVideo,
+			wantModelID:  "streamdiffusion",
+		},
+		{
+			name: "byoc: oversized override labels are length-capped",
+			req: RemotePaymentRequest{
+				Type:       RemoteType_LiveVideoToVideo,
+				Capability: strings.Repeat("c", maxUsageLabelLen+50),
+				ModelID:    strings.Repeat("m", maxUsageLabelLen+50),
+			},
+			caps:         nil,
+			wantPipeline: strings.Repeat("c", maxUsageLabelLen),
+			wantModelID:  strings.Repeat("m", maxUsageLabelLen),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			require := require.New(t)
+			req := tc.req
+			pipeline, modelID := resolveUsageLabels(&req, tc.caps)
+			require.Equal(tc.wantPipeline, pipeline, "pipeline")
+			require.Equal(tc.wantModelID, modelID, "model_id")
+		})
+	}
+}