From b67d7d30fac11f46c58523fa5ab8917d8e546d84 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Sat, 13 Jun 2026 02:18:42 -0400 Subject: [PATCH 1/6] feat(capabilities): Add ModelIDForCapability function and corresponding tests - Implemented the ModelIDForCapability method in the Capabilities struct to return the constrained model ID for a given capability, ensuring deterministic results when multiple models are present. - Added unit tests for ModelIDForCapability to cover various scenarios, including nil receivers, no constraints, single constrained models, and handling of ambiguous model cases. - Updated the remote signer to include the model ID in the event payload for LiveVideoToVideo capabilities. --- core/capabilities.go | 25 ++++++++++++++++++ core/capabilities_test.go | 54 +++++++++++++++++++++++++++++++++++++++ server/remote_signer.go | 3 +++ 3 files changed, 82 insertions(+) diff --git a/core/capabilities.go b/core/capabilities.go index e1ecd62576..6bef2c5480 100644 --- a/core/capabilities.go +++ b/core/capabilities.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "sync" "github.com/Masterminds/semver/v3" @@ -791,6 +792,30 @@ 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] +} + func (bcast *Capabilities) SetMinVersionConstraint(minVersionConstraint string) { if bcast != nil { bcast.constraints.minVersion = minVersionConstraint diff --git a/core/capabilities_test.go b/core/capabilities_test.go index 59f45b0582..5eba55217c 100644 --- a/core/capabilities_test.go +++ b/core/capabilities_test.go @@ -751,6 +751,60 @@ 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 are ambiguous, so no model is returned + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{ + "comfyui": &ModelConstraint{}, + "streamdiffusion": &ModelConstraint{}, + "noop": &ModelConstraint{}, + }, + }, + }) + assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) +} + func (c *Constraints) addCapabilityConstraints(cap Capability, constraint CapabilityConstraints) { // the capability should be added by AddCapacity for modelID, modelConstraint := range constraint.Models { diff --git a/server/remote_signer.go b/server/remote_signer.go index fd926cf574..2c167150ce 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -681,14 +681,17 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req sessionStatus = "new" } pipeline := "" + modelID := "" if req.Type == RemoteType_LiveVideoToVideo { pipeline = PipelineLiveVideoToVideo + modelID = streamParams.Capabilities.ModelIDForCapability(core.Capability_LiveVideoToVideo) } // 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, From e56beff30e9de1d982a5ab1ec8ab62d6c70c3b4a Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Thu, 25 Jun 2026 17:16:09 -0400 Subject: [PATCH 2/6] fix(capabilities): Update ModelIDForCapability test for lexicographical resolution - Modified the test for ModelIDForCapability to reflect that multiple models resolve to the lexicographically first ID instead of returning an empty string. - Adjusted the expected output in the test case to return "comfyui" for the Capability_LiveVideoToVideo capability. --- core/capabilities_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/capabilities_test.go b/core/capabilities_test.go index 5eba55217c..7f602dd98b 100644 --- a/core/capabilities_test.go +++ b/core/capabilities_test.go @@ -792,7 +792,7 @@ func TestModelIDForCapability(t *testing.T) { }) assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo)) - // multiple models are ambiguous, so no model is returned + // multiple models resolve to the lexicographically first ID c.SetPerCapabilityConstraints(PerCapabilityConstraints{ Capability_LiveVideoToVideo: &CapabilityConstraints{ Models: ModelConstraints{ @@ -802,7 +802,7 @@ func TestModelIDForCapability(t *testing.T) { }, }, }) - assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + assert.Equal("comfyui", c.ModelIDForCapability(Capability_LiveVideoToVideo)) } func (c *Constraints) addCapabilityConstraints(cap Capability, constraint CapabilityConstraints) { From aa9189b44fd605387b6a20f3b5db671f6bd518ee Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Thu, 25 Jun 2026 17:23:36 -0400 Subject: [PATCH 3/6] ci: trigger docker build for PR #3947 From b88305a37963e21dfa0a166d4840fbf6eca4570a Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Fri, 26 Jun 2026 13:30:44 -0400 Subject: [PATCH 4/6] Enhance remote signer payment handling with auth ID support - Updated the payment processing logic to include an optional `AuthID` in the `RemotePaymentState` and `authResponse` structures. - Modified the `doPaymentWithStateAndAuthID` function to accept and set the `AuthID` from the request header or callback response. - Added tests to verify the correct handling of `AuthID` in various scenarios, including fallback behavior and error handling when the `AuthID` changes unexpectedly. - Ensured that the `AuthID` is included in the response payload for better tracking and debugging. --- server/remote_signer.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/remote_signer.go b/server/remote_signer.go index 2c167150ce..e66cf9f390 100644 --- a/server/remote_signer.go +++ b/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"` } From cfe860a249e70976037e0fc0ad3e9d0ce33c24de Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Fri, 26 Jun 2026 22:15:20 -0400 Subject: [PATCH 5/6] ci(docker): rebuild PR images without shared registry cache Shared livepeerci/build:cache caused PR Docker tags (pr-3947, feat-add-model-id-signer-kafka) to point at an older livepeer binary while metadata listed the PR head SHA. Skip registry cache on pull_request builds and pass GIT_REVISION to bust the compile layer. --- .github/workflows/docker.yaml | 7 +++++-- docker/Dockerfile | 3 +++ 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index 5ecfc8379c..56813d0d07 100644 --- a/.github/workflows/docker.yaml +++ b/.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 diff --git a/docker/Dockerfile b/docker/Dockerfile index efa33a4d47..1ae4ed8032 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -61,6 +61,9 @@ COPY ./install_ffmpeg.sh ./install_ffmpeg.sh 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 From 9754c283a70eb6e8a317ec9c6a6f08625f861499 Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Wed, 8 Jul 2026 16:09:45 -0400 Subject: [PATCH 6/6] feat(capabilities): Add capability to pipeline conversion and constrained model ID resolution - Introduced `CapabilityToPipeline` function to convert capabilities to their corresponding pipeline slugs, returning an empty string for unknown capabilities. - Implemented `ConstrainedPipelineModelID` method to return a deterministic pipeline/model ID pair based on per-capability constraints, ensuring lexicographical resolution when multiple capabilities are present. - Added unit tests for both new functions to validate their behavior under various scenarios, including handling of nil capabilities and multiple constraints. - Updated the remote signer to utilize the new constrained pipeline logic for improved payment processing. --- core/capabilities.go | 51 +++++++++++++++++++++++++++++++++++++++ core/capabilities_test.go | 50 ++++++++++++++++++++++++++++++++++++++ server/remote_signer.go | 6 +++-- 3 files changed, 105 insertions(+), 2 deletions(-) diff --git a/core/capabilities.go b/core/capabilities.go index 6bef2c5480..838ede18f5 100644 --- a/core/capabilities.go +++ b/core/capabilities.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sort" + "strings" "sync" "github.com/Masterminds/semver/v3" @@ -816,6 +817,56 @@ func (bcast *Capabilities) ModelIDForCapability(cap Capability) string { 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 diff --git a/core/capabilities_test.go b/core/capabilities_test.go index 7f602dd98b..db3f0a6e61 100644 --- a/core/capabilities_test.go +++ b/core/capabilities_test.go @@ -805,6 +805,56 @@ func TestModelIDForCapability(t *testing.T) { 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 { diff --git a/server/remote_signer.go b/server/remote_signer.go index e66cf9f390..f34cdead52 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -682,9 +682,11 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req } pipeline := "" modelID := "" - if req.Type == RemoteType_LiveVideoToVideo { + if streamParams.Capabilities != nil { + pipeline, modelID = streamParams.Capabilities.ConstrainedPipelineModelID() + } + if pipeline == "" && req.Type == RemoteType_LiveVideoToVideo { pipeline = PipelineLiveVideoToVideo - modelID = streamParams.Capabilities.ModelIDForCapability(core.Capability_LiveVideoToVideo) } // NB: This could could drop events if tha Kafka queue is full! monitor.SendQueueEventAsync("create_signed_ticket", map[string]interface{}{