feat(remote_signer): emit model_id in create_signed_ticket Kafka events

.github/workflows/docker.yaml

@@ -82,6 +82,7 @@ jobs:
         with:
           build-args: |
             BUILD_TAGS=${{ steps.build-tag.outputs.build-tags }}
+            GIT_REVISION=${{ github.event.pull_request.head.sha || github.sha }}
           context: .
           provenance: mode=max
           sbom: true
@@ -92,8 +93,10 @@ jobs:
           tags: ${{ steps.meta.outputs.tags }}
           annotations: ${{ steps.meta.outputs.annotations }}
           labels: ${{ steps.meta.outputs.labels }}
-          cache-from: type=registry,ref=livepeerci/build:cache
-          cache-to: type=registry,ref=livepeerci/build:cache,mode=max
+          # PR builds must not reuse the shared registry cache — it can serve a
+          # stale livepeer binary while metadata tags the PR head SHA (see #3947).
+          cache-from: ${{ github.event_name != 'pull_request' && 'type=registry,ref=livepeerci/build:cache' || '' }}
+          cache-to: ${{ github.event_name != 'pull_request' && 'type=registry,ref=livepeerci/build:cache,mode=max' || '' }}
 
   builder:
     name: go-livepeer builder docker image generation

core/capabilities.go

@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
+	"strings"
 	"sync"
 
 	"github.com/Masterminds/semver/v3"
@@ -791,6 +793,80 @@ func (bcast *Capabilities) PerCapability() PerCapabilityConstraints {
 	return nil
 }
 
+// ModelIDForCapability returns the constrained model ID for the given capability.
+// When multiple models are constrained it returns the lexicographically first one
+// so the result is deterministic. Returns "" when no model is constrained.
+func (bcast *Capabilities) ModelIDForCapability(cap Capability) string {
+	if bcast == nil {
+		return ""
+	}
+	capConstraints, ok := bcast.constraints.perCapability[cap]
+	if !ok || capConstraints == nil {
+		return ""
+	}
+	modelIDs := make([]string, 0, len(capConstraints.Models))
+	for modelID := range capConstraints.Models {
+		if modelID != "" {
+			modelIDs = append(modelIDs, modelID)
+		}
+	}
+	if len(modelIDs) == 0 {
+		return ""
+	}
+	sort.Strings(modelIDs)
+	return modelIDs[0]
+}
+
+// CapabilityToPipeline converts a capability to a pipeline slug.
+// Returns "" when the capability is unknown.
+func CapabilityToPipeline(cap Capability) string {
+	capName, ok := CapabilityNameLookup[cap]
+	if !ok {
+		return ""
+	}
+	return strings.ReplaceAll(strings.ToLower(capName), " ", "-")
+}
+
+// ConstrainedPipelineModelID returns a deterministic pipeline/model_id pair
+// from per-capability constraints. When multiple capabilities have constrained
+// models, it returns the lexicographically first pipeline.
+func (bcast *Capabilities) ConstrainedPipelineModelID() (string, string) {
+	if bcast == nil {
+		return "", ""
+	}
+
+	type capPipeline struct {
+		capability Capability
+		pipeline   string
+	}
+
+	candidates := []capPipeline{}
+	for capability := range bcast.constraints.perCapability {
+		modelID := bcast.ModelIDForCapability(capability)
+		if modelID == "" {
+			continue
+		}
+		pipeline := CapabilityToPipeline(capability)
+		if pipeline == "" {
+			continue
+		}
+		candidates = append(candidates, capPipeline{
+			capability: capability,
+			pipeline:   pipeline,
+		})
+	}
+	if len(candidates) == 0 {
+		return "", ""
+	}
+
+	sort.Slice(candidates, func(i, j int) bool {
+		return candidates[i].pipeline < candidates[j].pipeline
+	})
+
+	selected := candidates[0]
+	return selected.pipeline, bcast.ModelIDForCapability(selected.capability)
+}
+
 func (bcast *Capabilities) SetMinVersionConstraint(minVersionConstraint string) {
 	if bcast != nil {
 		bcast.constraints.minVersion = minVersionConstraint

core/capabilities_test.go

@@ -751,6 +751,110 @@ func TestMinRunnerVersion(t *testing.T) {
 	assert.Equal("", c.MinRunnerVersionConstraint(Capability_LiveVideoToVideo, "other"))
 }
 
+func TestModelIDForCapability(t *testing.T) {
+	assert := assert.New(t)
+
+	// nil receiver is safe
+	var nilCaps *Capabilities
+	assert.Equal("", nilCaps.ModelIDForCapability(Capability_LiveVideoToVideo))
+
+	// no constraints for the capability
+	c := NewCapabilities([]Capability{Capability_LiveVideoToVideo}, nil)
+	assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo))
+
+	// single constrained model
+	c.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_LiveVideoToVideo: &CapabilityConstraints{
+			Models: ModelConstraints{"streamdiffusion": &ModelConstraint{Warm: true}},
+		},
+	})
+	assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo))
+
+	// different capability has no constraint
+	assert.Equal("", c.ModelIDForCapability(Capability_TextToImage))
+
+	// empty model IDs are ignored
+	c.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_LiveVideoToVideo: &CapabilityConstraints{
+			Models: ModelConstraints{"": &ModelConstraint{}},
+		},
+	})
+	assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo))
+
+	// one real model plus empty key resolves to the real model
+	c.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_LiveVideoToVideo: &CapabilityConstraints{
+			Models: ModelConstraints{
+				"":                &ModelConstraint{},
+				"streamdiffusion": &ModelConstraint{Warm: true},
+			},
+		},
+	})
+	assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo))
+
+	// multiple models resolve to the lexicographically first ID
+	c.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_LiveVideoToVideo: &CapabilityConstraints{
+			Models: ModelConstraints{
+				"comfyui":         &ModelConstraint{},
+				"streamdiffusion": &ModelConstraint{},
+				"noop":            &ModelConstraint{},
+			},
+		},
+	})
+	assert.Equal("comfyui", c.ModelIDForCapability(Capability_LiveVideoToVideo))
+}
+
+func TestCapabilityToPipeline(t *testing.T) {
+	assert := assert.New(t)
+
+	assert.Equal("live-video-to-video", CapabilityToPipeline(Capability_LiveVideoToVideo))
+	assert.Equal("text-to-image", CapabilityToPipeline(Capability_TextToImage))
+	assert.Equal("", CapabilityToPipeline(Capability(-999)))
+}
+
+func TestConstrainedPipelineModelID(t *testing.T) {
+	assert := assert.New(t)
+
+	var nilCaps *Capabilities
+	pipeline, modelID := nilCaps.ConstrainedPipelineModelID()
+	assert.Equal("", pipeline)
+	assert.Equal("", modelID)
+
+	c := NewCapabilities([]Capability{Capability_LiveVideoToVideo, Capability_TextToImage}, nil)
+	pipeline, modelID = c.ConstrainedPipelineModelID()
+	assert.Equal("", pipeline)
+	assert.Equal("", modelID)
+
+	c.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_TextToImage: &CapabilityConstraints{
+			Models: ModelConstraints{
+				"stabilityai/sd-turbo": &ModelConstraint{Warm: true},
+			},
+		},
+	})
+	pipeline, modelID = c.ConstrainedPipelineModelID()
+	assert.Equal("text-to-image", pipeline)
+	assert.Equal("stabilityai/sd-turbo", modelID)
+
+	// Multiple constrained capabilities resolve deterministically.
+	c.SetPerCapabilityConstraints(PerCapabilityConstraints{
+		Capability_LiveVideoToVideo: &CapabilityConstraints{
+			Models: ModelConstraints{
+				"streamdiffusion": &ModelConstraint{},
+			},
+		},
+		Capability_AudioToText: &CapabilityConstraints{
+			Models: ModelConstraints{
+				"whisper-large": &ModelConstraint{},
+			},
+		},
+	})
+	pipeline, modelID = c.ConstrainedPipelineModelID()
+	assert.Equal("audio-to-text", pipeline)
+	assert.Equal("whisper-large", modelID)
+}
+
 func (c *Constraints) addCapabilityConstraints(cap Capability, constraint CapabilityConstraints) {
 	// the capability should be added by AddCapacity
 	for modelID, modelConstraint := range constraint.Models {

docker/Dockerfile

@@ -1,4 +1,4 @@
 FROM --platform=$BUILDPLATFORM livepeerci/cuda:12.0.0-cudnn8-devel-ubuntu20.04 as build
 
 ARG	TARGETARCH
 ARG	BUILDARCH
@@ -61,6 +61,9 @@
 ARG	BUILD_TAGS
 ENV	BUILD_TAGS=${BUILD_TAGS}
 
+ARG	GIT_REVISION=unknown
+RUN	echo "Building livepeer at ${GIT_REVISION}"
+
 COPY	go.mod	go.sum	./
 RUN	go mod download
 
@@ -72,9 +75,9 @@
 
 RUN	make livepeer livepeer_cli livepeer_bench livepeer_router
 
 FROM --platform=$TARGETPLATFORM	nvidia/cuda:12.0.0-cudnn8-runtime-ubuntu20.04 AS livepeer-amd64-base
 
 FROM --platform=$TARGETPLATFORM	nvidia/cuda:12.0.0-cudnn8-runtime-ubuntu20.04 AS livepeer-arm64-base
 
 FROM	livepeer-${TARGETARCH}-base
 

server/remote_signer.go

@@ -269,7 +269,7 @@ type authResponse struct {
 	// Unix timestamp (seconds) until which auth is considered valid.
 	// Allows for skipping webhook callbacks until this time is exceeded.
 	Expiry int64 `json:"expiry,omitempty"`
-	// Optional opaque identifier.
+	// Optional opaque identifier for metering attribution.
 	AuthID string `json:"auth_id,omitempty"`
 }
 
@@ -681,14 +681,19 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 			sessionStatus = "new"
 		}
 		pipeline := ""
-		if req.Type == RemoteType_LiveVideoToVideo {
+		modelID := ""
+		if streamParams.Capabilities != nil {
+			pipeline, modelID = streamParams.Capabilities.ConstrainedPipelineModelID()
+		}
+		if pipeline == "" && req.Type == RemoteType_LiveVideoToVideo {
 			pipeline = PipelineLiveVideoToVideo
 		}
 		// NB: This could could drop events if tha Kafka queue is full!
 		monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{
 			"session_id":         state.StateID,
 			"session_status":     sessionStatus,
 			"pipeline":           pipeline,
+			"model_id":           modelID,
 			"request_id":         requestID,
 			"orch_address":       orchAddr.Hex(),
 			"orch_url":           oInfo.Transcoder,