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,32 @@ func (ls *LivepeerServer) authLivePayment(r *http.Request, state *RemotePaymentS
 	return *webhookResp.Status, &webhookResp, errors.New(webhookResp.Reason)
 }
 
+// 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 so BYOC jobs are attributed to their true
+// pipeline + model. 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 both remain empty (the collector then defaults them to
+// "unknown"). This makes non-BYOC (lv2v) callers byte-identical to before and
+// keeps usage attribution decoupled from `Type`, which still drives fee/pixel
+// routing.
+func resolveUsageLabels(req *RemotePaymentRequest, caps *core.Capabilities) (pipeline, modelID string) {
+	if req.Type == RemoteType_LiveVideoToVideo {
+		pipeline = PipelineLiveVideoToVideo
+		modelID = caps.ModelIDForCapability(core.Capability_LiveVideoToVideo)
+	}
+	if req.Capability != "" {
+		pipeline = req.Capability
+	}
+	if req.ModelID != "" {
+		modelID = req.ModelID
+	}
+	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 +720,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

@@ -1692,3 +1692,89 @@ 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) {
+	require := require.New(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:  "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			req := tc.req
+			pipeline, modelID := resolveUsageLabels(&req, tc.caps)
+			require.Equal(tc.wantPipeline, pipeline, "pipeline")
+			require.Equal(tc.wantModelID, modelID, "model_id")
+		})
+	}
+}