feat(playback): plan v3 routes from pooled node capabilities (#408)

* feat(playback): plan v3 routes from pooled node capabilities Protocol v3 planning previously gated every server transformation on the API host's local ffmpeg probe, so deployments whose toolchain lives on transcode nodes (libx264/aac/dovi_rpu on nodes, minimal binary locally) received conversion terminals before transport preparation ever consulted the selected node's capabilities. Planning now draws on two registries split by executor pool: - Registry stays the local probe and keeps gating progressive remux routes, which execute in this process and can never offload. - HLSRegistry widens availability for HLS deliveries with the pooled transcode nodes' advertised transformations (name and recipe version pinned to the local specs), fetched concurrently under a short planning deadline through the existing TTL cache. Failures are now negatively cached so an unreachable node costs one timeout per window rather than one per start. The remux family picks the executor per branch: a recipe needing transformations only nodes carry skips the progressive remux and ships the same recipe on the HLS remux delivery instead. The local-fallback path in prepareTransportV3 now validates plans against the local registry's advertised set — mirroring the per-node validation — and returns the existing retryable transcode_node_capability_unavailable terminal when no executor can run the recipe, instead of spawning an ffmpeg that would fail at runtime. Deferred from PR #398 review (comment 3579105380). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(playback): harden union capability planning from review Addresses all four review findings on the capability-union feature: - Select capability-matching nodes: plans carrying server transformations now restrict node selection to nodes whose advertised capabilities validate against the plan (nodepool.PlanSessionWith with a set-lookup predicate), so heterogeneous pools cannot load-balance a recipe onto a node that would reject it while a capable sibling exists. Transformation-free plans keep pure load-based selection. - Split the capability cache by consumer: planning honors negatively cached fetch failures (one timeout per window), while the transport path fetches through them — a memoized 3s planning deadline must not reject an already-selected node that the 10s transport budget could still validate. - Gate node-widened availability on the HLS engine: a progressive-only client that needs audio conversion keeps its specific retryable audio_conversion_unsupported terminal instead of falling through to a non-retryable adaptation_unavailable for routes it can never run; the DV strip union flag is gated identically. - Make HLSRegistry a lazy, memoized producer: the planner only builds the widened registry when a route decision depends on node capabilities, so direct-play and other source-preserving starts never wait on node capability fetches (or their dead-node deadlines). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 12:13:16 -04:00
parent 854d07cf8f
commit 075e217477
8 changed files with 746 additions and 21 deletions
@@ -35,10 +35,18 @@ const (
 	maxPlaybackV3EventBodyBytes = 32 << 10
 	replanLeaseDurationV3       = 15 * time.Second
 	v3NodeCapabilityTTL         = time.Minute
+	// Failed capability fetches are memoized briefly so an unreachable node
+	// costs one timeout per window instead of one per planning request.
+	v3NodeCapabilityErrorTTL = 15 * time.Second
+	// Capability fetches on the planning path run under a deadline well below
+	// the fetch helper's own 10s timeout: planning happens on the start
+	// request path, where a slow node must degrade the union, not the user.
+	v3NodeCapabilityPlanTimeout = 3 * time.Second
 )

 type v3NodeCapabilityCache struct {
 	transformations []playback.TransformationV3
+	err             error
 	expiresAt       time.Time
 }

@@ -139,17 +147,44 @@ func (h *PlaybackHandler) transformationRegistryV3(ctx context.Context) *playbac
 	return h.v3Registry
 }

+// remoteTransformationsV3 is the transport-time capability lookup for a
+// selected node. It never trusts memoized failures: those may be planning
+// deadlines far shorter than this path's fetch budget, and rejecting the
+// already-selected node on a stale planning timeout would fail a start the
+// fetch could still validate.
 func (h *PlaybackHandler) remoteTransformationsV3(ctx context.Context, nodeURL string) ([]playback.TransformationV3, error) {
+	return h.lookupRemoteTransformationsV3(ctx, nodeURL, false)
+}
+
+// remoteTransformationsPlanningV3 is the planning-time variant: it honors
+// negatively-cached fetch failures so an unreachable node costs one timeout
+// per error-TTL window instead of one per playback start.
+func (h *PlaybackHandler) remoteTransformationsPlanningV3(ctx context.Context, nodeURL string) ([]playback.TransformationV3, error) {
+	return h.lookupRemoteTransformationsV3(ctx, nodeURL, true)
+}
+
+func (h *PlaybackHandler) lookupRemoteTransformationsV3(ctx context.Context, nodeURL string, honorCachedFailure bool) ([]playback.TransformationV3, error) {
 	now := time.Now()
 	h.v3NodeCapabilitiesMu.Lock()
 	entry, ok := h.v3NodeCapabilities[nodeURL]
 	h.v3NodeCapabilitiesMu.Unlock()
 	if ok && now.Before(entry.expiresAt) {
-		return append([]playback.TransformationV3(nil), entry.transformations...), nil
+		if entry.err == nil {
+			return append([]playback.TransformationV3(nil), entry.transformations...), nil
+		}
+		if honorCachedFailure {
+			return nil, entry.err
+		}
 	}

 	info, err := fetchRemoteTranscodeCapabilities(ctx, nodeURL, h.JWTSecret)
 	if err != nil {
+		h.v3NodeCapabilitiesMu.Lock()
+		if h.v3NodeCapabilities == nil {
+			h.v3NodeCapabilities = make(map[string]v3NodeCapabilityCache)
+		}
+		h.v3NodeCapabilities[nodeURL] = v3NodeCapabilityCache{err: err, expiresAt: now.Add(v3NodeCapabilityErrorTTL)}
+		h.v3NodeCapabilitiesMu.Unlock()
 		return nil, err
 	}
 	entry = v3NodeCapabilityCache{
@@ -165,7 +200,122 @@ func (h *PlaybackHandler) remoteTransformationsV3(ctx context.Context, nodeURL s
 	return append([]playback.TransformationV3(nil), entry.transformations...), nil
 }

-func validateRemoteTransformationsV3(plan *playback.PlanV3, advertised []playback.TransformationV3) error {
+// transcodeNodeEnumeratorV3 exposes the pooled transcode nodes whose
+// advertised transformations widen HLS planning; *nodepool.Planner implements
+// it.
+type transcodeNodeEnumeratorV3 interface {
+	TranscodeNodeURLs() []string
+}
+
+// hlsPlanningRegistryV3 returns the registry HLS deliveries plan against: the
+// local probe plus every pooled transcode node's advertised transformations.
+// Only availability of locally-defined specs widens (name and recipe version
+// pinned by this server), so any plan built from it passes the per-node
+// advertisement validation when that node is selected, and the local-fallback
+// validation in prepareTransportV3 rejects recipes only nodes can run.
+// Without pooled nodes this is exactly the local registry.
+func (h *PlaybackHandler) hlsPlanningRegistryV3(ctx context.Context) *playback.TransformationRegistryV3 {
+	local := h.transformationRegistryV3(ctx)
+	enumerator, ok := h.NodePlanner.(transcodeNodeEnumeratorV3)
+	if !ok {
+		return local
+	}
+	nodeURLs := enumerator.TranscodeNodeURLs()
+	if len(nodeURLs) == 0 {
+		return local
+	}
+	var merged []playback.TransformationV3
+	for _, transformations := range h.pooledNodeTransformationsV3(ctx, nodeURLs) {
+		merged = append(merged, transformations...)
+	}
+	return local.WithAdvertised(merged)
+}
+
+// lazyHLSPlanningRegistryV3 defers (and memoizes) the widened-registry build
+// so the planner only pays for node capability lookups when a route decision
+// actually depends on them; direct-play and other source-preserving starts
+// never touch the pool.
+func (h *PlaybackHandler) lazyHLSPlanningRegistryV3(ctx context.Context) func() *playback.TransformationRegistryV3 {
+	var once sync.Once
+	var registry *playback.TransformationRegistryV3
+	return func() *playback.TransformationRegistryV3 {
+		once.Do(func() { registry = h.hlsPlanningRegistryV3(ctx) })
+		return registry
+	}
+}
+
+// pooledNodeTransformationsV3 collects the advertised transformations of the
+// given transcode nodes, keyed by node URL. Stale cache entries are refreshed
+// concurrently under a short planning deadline; nodes that cannot be reached
+// contribute nothing (their failures are negatively cached), so planning
+// degrades toward the local registry instead of blocking the start path.
+func (h *PlaybackHandler) pooledNodeTransformationsV3(ctx context.Context, nodeURLs []string) map[string][]playback.TransformationV3 {
+	fetchCtx, cancel := context.WithTimeout(ctx, v3NodeCapabilityPlanTimeout)
+	defer cancel()
+	results := make([][]playback.TransformationV3, len(nodeURLs))
+	var wg sync.WaitGroup
+	for i, nodeURL := range nodeURLs {
+		wg.Add(1)
+		go func(i int, nodeURL string) {
+			defer wg.Done()
+			transformations, err := h.remoteTransformationsPlanningV3(fetchCtx, nodeURL)
+			if err != nil {
+				slog.DebugContext(ctx, "protocol v3 node capability unavailable for planning", "component", "api", "node", nodeURL, "error", err)
+				return
+			}
+			results[i] = transformations
+		}(i, nodeURL)
+	}
+	wg.Wait()
+	byURL := make(map[string][]playback.TransformationV3, len(nodeURLs))
+	for i, transformations := range results {
+		if transformations != nil {
+			byURL[nodeURLs[i]] = transformations
+		}
+	}
+	return byURL
+}
+
+// capabilitySessionPlannerV3 is implemented by *nodepool.Planner; it lets the
+// transport layer restrict node selection to nodes that can execute the
+// plan's server transformations.
+type capabilitySessionPlannerV3 interface {
+	PlanSessionWith(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int, eligible func(*nodepool.Node) bool) nodepool.Plan
+}
+
+// planNodeSessionV3 selects transcode/proxy nodes for the session. Plans that
+// carry server transformations restrict selection to nodes whose advertised
+// capabilities validate against the plan, so load balancing in a
+// heterogeneous pool cannot land a recipe on a node that would reject it when
+// a capable sibling exists. Capability-blind selection remains for
+// transformation-free plans and non-enumerating planners.
+func (h *PlaybackHandler) planNodeSessionV3(ctx context.Context, session *playback.Session, result playback.PlannerResultV3) nodepool.Plan {
+	selector, selectable := h.NodePlanner.(capabilitySessionPlannerV3)
+	enumerator, enumerable := h.NodePlanner.(transcodeNodeEnumeratorV3)
+	if !selectable || !enumerable || !planRequiresServerTransformationsV3(result.Plan) {
+		return h.NodePlanner.PlanSession(session.ID, session.TranscodeNodeURL, true, result.TargetBitrateKbps)
+	}
+	capable := make(map[string]struct{})
+	for nodeURL, advertised := range h.pooledNodeTransformationsV3(ctx, enumerator.TranscodeNodeURLs()) {
+		if validateAdvertisedTransformationsV3(result.Plan, advertised) == nil {
+			capable[nodeURL] = struct{}{}
+		}
+	}
+	// The predicate runs under the planner lock: a set lookup only.
+	return selector.PlanSessionWith(session.ID, session.TranscodeNodeURL, true, result.TargetBitrateKbps, func(node *nodepool.Node) bool {
+		if node == nil {
+			return false
+		}
+		_, ok := capable[node.URL]
+		return ok
+	})
+}
+
+// validateAdvertisedTransformationsV3 verifies that every server-executed
+// transformation the plan requires is advertised — at the exact recipe
+// version — by the executor under consideration (a pooled node's capability
+// response or the local registry's Advertised set).
+func validateAdvertisedTransformationsV3(plan *playback.PlanV3, advertised []playback.TransformationV3) error {
 	available := make(map[string]string, len(advertised))
 	for _, transformation := range advertised {
 		available[strings.ToLower(strings.TrimSpace(transformation.Name))] = strings.TrimSpace(transformation.RecipeVersion)
@@ -179,7 +329,7 @@ func validateRemoteTransformationsV3(plan *playback.PlanV3, advertised []playbac
 		}
 		version, ok := available[strings.ToLower(strings.TrimSpace(required.Name))]
 		if !ok || version != strings.TrimSpace(required.RecipeVersion) {
-			return fmt.Errorf("transcode node lacks transformation %s@%s", required.Name, required.RecipeVersion)
+			return fmt.Errorf("executor lacks transformation %s@%s", required.Name, required.RecipeVersion)
 		}
 	}
 	return nil
@@ -275,7 +425,7 @@ func (h *PlaybackHandler) handleStartPlaybackV3(w http.ResponseWriter, r *http.R
 	result := playback.PlanPlaybackV3(playback.PlannerInputV3{
 		Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile,
 		AudioTrackIndex: audioIndex, Settings: settings,
-		Registry: h.transformationRegistryV3(r.Context()), Now: time.Now(),
+		Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(),
 		AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile),
 	})
 	if result.Terminal != nil && result.Terminal.Reason == "no_alternate_version" && shouldTryAlternateFileV3(req.QualityPreference) {
@@ -290,7 +440,7 @@ func (h *PlaybackHandler) handleStartPlaybackV3(w http.ResponseWriter, r *http.R
 				writePlaybackFilePreflightError(w, err)
 				return
 			}
-			result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: settings, Registry: h.transformationRegistryV3(r.Context()), Now: time.Now(), AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
+			result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: settings, Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
 		}
 	}
 	if result.Terminal != nil {
@@ -401,11 +551,11 @@ func (h *PlaybackHandler) prepareTransportV3(r *http.Request, session *playback.
 		return h.prepareIdentityTransportV3(session, result), nil
 	}
 	if h.NodePlanner != nil {
-		plan := h.NodePlanner.PlanSession(session.ID, session.TranscodeNodeURL, true, result.TargetBitrateKbps)
+		plan := h.planNodeSessionV3(r.Context(), session, result)
 		if plan.TranscodeNode != nil {
 			transformations, err := h.remoteTransformationsV3(r.Context(), plan.TranscodeNode.URL)
 			if err == nil {
-				err = validateRemoteTransformationsV3(result.Plan, transformations)
+				err = validateAdvertisedTransformationsV3(result.Plan, transformations)
 			}
 			if err == nil {
 				transport, transportErr := h.prepareRemoteTransportV3(r, session, file, result, plan)
@@ -428,9 +578,34 @@ func (h *PlaybackHandler) prepareTransportV3(r *http.Request, session *playback.
 			return preparedTransportV3{}, &transportErrorV3{reason: "capacity_unavailable", message: "No transcode node is available and local fallback is disabled.", retryable: true}
 		}
 	}
+	// Capability-union planning may select transformations only pooled nodes
+	// can execute; the local binary must prove it carries the recipe before
+	// this fallback spawns an ffmpeg that would fail at runtime. Retryable:
+	// a capable node freeing up satisfies the same plan. Transformation-free
+	// plans skip the check (and the local probe behind it) entirely.
+	if planRequiresServerTransformationsV3(result.Plan) {
+		if err := validateAdvertisedTransformationsV3(result.Plan, h.transformationRegistryV3(r.Context()).Advertised()); err != nil {
+			return preparedTransportV3{}, &transportErrorV3{reason: "transcode_node_capability_unavailable", message: "No available transcode executor can run the selected playback recipe.", retryable: true, cause: err}
+		}
+	}
 	return h.prepareLocalTransportV3(r, session, file, result)
 }

+// planRequiresServerTransformationsV3 reports whether the plan carries any
+// transformation the serving executor (local binary or transcode node) must
+// perform, as opposed to client-executed ones.
+func planRequiresServerTransformationsV3(plan *playback.PlanV3) bool {
+	if plan == nil {
+		return false
+	}
+	for _, transformation := range plan.Transformations {
+		if !strings.EqualFold(transformation.Executor, "client") {
+			return true
+		}
+	}
+	return false
+}
+
 func (h *PlaybackHandler) prepareIdentityTransportV3(session *playback.Session, result playback.PlannerResultV3) preparedTransportV3 {
 	routeSession := *session
 	routeSession.PlayMethod = result.PlayMethod
@@ -1014,7 +1189,7 @@ func (h *PlaybackHandler) executeReplanV3(r *http.Request, record *playback.Atte
 			attemptedKeys = append(attemptedKeys, currentKey)
 		}
 	}
-	result := playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
+	result := playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
 	if result.Terminal != nil && result.Terminal.Reason == "no_alternate_version" && replanAllowsAlternateFileV3(operation, start.QualityPreference) {
 		if alternate, alternateErr := h.findAlternateFile(r.Context(), requestedFile); alternateErr == nil && alternate != nil {
 			alternate = h.ensurePlaybackProbe(r.Context(), alternate)
@@ -1024,7 +1199,7 @@ func (h *PlaybackHandler) executeReplanV3(r *http.Request, record *playback.Atte
 				if err := preflightPlaybackFile(r.Context(), alternate, h.MissingMarker, h.EventsHub); err == nil {
 					effectiveFile = alternate
 					audioIndex = remappedAudio
-					result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
+					result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)})
 				}
 			}
 		}
@@ -43,6 +43,7 @@ func (h *PlaybackHandler) shadowLegacyPlaybackV3(ctx context.Context, req startP
 		AudioTrackIndex: audioIndex,
 		Settings:        h.plannerSettingsV3(ctx),
 		Registry:        h.transformationRegistryV3(ctx),
+		HLSRegistry:     h.lazyHLSPlanningRegistryV3(ctx),
 		Now:             time.Now(),
 	})
 	attrs := []any{
@@ -0,0 +1,198 @@
+package handlers
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/Silo-Server/silo-server/internal/nodepool"
+	"github.com/Silo-Server/silo-server/internal/playback"
+)
+
+// enumeratingNodePlannerV3 is a SessionPlanner stub that also exposes pooled
+// transcode node URLs, matching *nodepool.Planner's production shape.
+type enumeratingNodePlannerV3 struct {
+	staticNodePlannerV3
+	urls []string
+}
+
+func (p enumeratingNodePlannerV3) TranscodeNodeURLs() []string { return p.urls }
+
+// presetLocalRegistryV3 pins the handler's local transformation registry so
+// tests never probe the machine's real ffmpeg.
+func presetLocalRegistryV3(h *PlaybackHandler, registry *playback.TransformationRegistryV3) {
+	h.v3RegistryOnce.Do(func() {})
+	h.v3Registry = registry
+}
+
+func TestHLSPlanningRegistryV3UnionsPooledNodeCapabilities(t *testing.T) {
+	remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/hw-capabilities" {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		writeJSON(w, http.StatusOK, playback.HWAccelInfo{Transformations: []playback.TransformationV3{
+			{Name: "video_to_h264", Executor: "server", RecipeVersion: "1"},
+			{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"},
+		}})
+	}))
+	defer remote.Close()
+
+	handler := NewPlaybackHandler(playback.NewSessionManager(0, 0))
+	handler.JWTSecret = "test-secret"
+	presetLocalRegistryV3(handler, playback.NewTransformationRegistryV3([]playback.TransformationSpecV3{
+		{Name: "video_to_h264", RecipeVersion: "1"},
+		{Name: "audio_to_aac", RecipeVersion: "1"},
+		{Name: "server_dv7_to_hdr10", RecipeVersion: "1"},
+	}))
+	handler.NodePlanner = enumeratingNodePlannerV3{urls: []string{remote.URL}}
+
+	registry := handler.hlsPlanningRegistryV3(context.Background())
+	if !registry.Available("video_to_h264") || !registry.Available("audio_to_aac") {
+		t.Fatal("pooled node capabilities must widen the HLS planning registry")
+	}
+	if registry.Available("server_dv7_to_hdr10") {
+		t.Fatal("transformations no node advertises must stay unavailable")
+	}
+	if handler.transformationRegistryV3(context.Background()).Available("video_to_h264") {
+		t.Fatal("the local registry must not be widened by node capabilities")
+	}
+}
+
+func TestHLSPlanningRegistryV3WithoutEnumeratorIsLocal(t *testing.T) {
+	handler := NewPlaybackHandler(playback.NewSessionManager(0, 0))
+	local := playback.NewTransformationRegistryV3([]playback.TransformationSpecV3{{Name: "audio_to_aac", RecipeVersion: "1", Available: true}})
+	presetLocalRegistryV3(handler, local)
+	handler.NodePlanner = staticNodePlannerV3{plan: nodepool.Plan{}}
+
+	if registry := handler.hlsPlanningRegistryV3(context.Background()); registry != local {
+		t.Fatal("a planner without node enumeration must plan from the local registry")
+	}
+}
+
+func TestRemoteTransformationsV3FailureCacheSplit(t *testing.T) {
+	hits := 0
+	fail := true
+	remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits++
+		if fail {
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		writeJSON(w, http.StatusOK, playback.HWAccelInfo{Transformations: []playback.TransformationV3{{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"}}})
+	}))
+	defer remote.Close()
+
+	handler := NewPlaybackHandler(playback.NewSessionManager(0, 0))
+	handler.JWTSecret = "test-secret"
+	if _, err := handler.remoteTransformationsPlanningV3(context.Background(), remote.URL); err == nil {
+		t.Fatal("fetch against a failing node must error")
+	}
+	if _, err := handler.remoteTransformationsPlanningV3(context.Background(), remote.URL); err == nil {
+		t.Fatal("planning lookups must surface the memoized failure")
+	}
+	if hits != 1 {
+		t.Fatalf("failing node was fetched %d times; planning must memoize the failure", hits)
+	}
+
+	// The transport path must fetch through the memoized failure: it may
+	// have been produced by a planning deadline far shorter than this
+	// path's budget, and rejecting the already-selected node on it would
+	// fail a start a fresh fetch could still validate.
+	fail = false
+	transformations, err := handler.remoteTransformationsV3(context.Background(), remote.URL)
+	if err != nil || len(transformations) != 1 {
+		t.Fatalf("transport lookup must refetch through a memoized failure: %v %#v", err, transformations)
+	}
+	if hits != 2 {
+		t.Fatalf("transport lookup fetched %d times, want 2", hits)
+	}
+	// The refetched success replaces the failure for planning too.
+	if _, err := handler.remoteTransformationsPlanningV3(context.Background(), remote.URL); err != nil {
+		t.Fatalf("planning lookup after transport success: %v", err)
+	}
+	if hits != 2 {
+		t.Fatalf("cached success was refetched (%d hits)", hits)
+	}
+}
+
+// In a heterogeneous pool, a plan that needs server transformations must be
+// placed on a node advertising them even when load balancing prefers an
+// incapable node, while transformation-free plans keep load-based selection.
+func TestPlanNodeSessionV3PrefersCapabilityMatchingNode(t *testing.T) {
+	capable := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		writeJSON(w, http.StatusOK, playback.HWAccelInfo{Transformations: []playback.TransformationV3{
+			{Name: "video_to_h264", Executor: "server", RecipeVersion: "1"},
+			{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"},
+		}})
+	}))
+	defer capable.Close()
+	incapable := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		writeJSON(w, http.StatusOK, playback.HWAccelInfo{})
+	}))
+	defer incapable.Close()
+
+	transcodes := nodepool.NewTranscodePool()
+	transcodes.SetNodes([]*nodepool.Node{
+		{ID: 1, Name: "incapable", Type: nodepool.NodeTypeTranscode, URL: incapable.URL, Enabled: true, Healthy: true, ActiveJobs: 0},
+		{ID: 2, Name: "capable", Type: nodepool.NodeTypeTranscode, URL: capable.URL, Enabled: true, Healthy: true, ActiveJobs: 5},
+	})
+	handler := NewPlaybackHandler(playback.NewSessionManager(0, 0))
+	handler.JWTSecret = "test-secret"
+	handler.NodePlanner = nodepool.NewPlanner(nodepool.NewProxyPool(), transcodes)
+
+	plan := &playback.PlanV3{
+		PlanID:   "plan:heterogeneous",
+		Delivery: playback.DeliveryTranscodeHLSV3,
+		Transformations: []playback.TransformationV3{
+			{Name: "video_to_h264", Executor: "server", RecipeVersion: "1"},
+			{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"},
+		},
+	}
+	selected := handler.planNodeSessionV3(context.Background(), &playback.Session{ID: "session-hetero"}, playback.PlannerResultV3{Plan: plan, PlayMethod: playback.PlayTranscode})
+	if selected.TranscodeNode == nil || selected.TranscodeNode.URL != capable.URL {
+		t.Fatalf("capability-requiring plan selected %+v, want the capable node", selected.TranscodeNode)
+	}
+
+	free := &playback.PlanV3{PlanID: "plan:copy", Delivery: playback.DeliveryRemuxHLSV3, Transformations: []playback.TransformationV3{}}
+	loadBased := handler.planNodeSessionV3(context.Background(), &playback.Session{ID: "session-copy"}, playback.PlannerResultV3{Plan: free, PlayMethod: playback.PlayRemux})
+	if loadBased.TranscodeNode == nil || loadBased.TranscodeNode.URL != incapable.URL {
+		t.Fatalf("transformation-free plan selected %+v, want load-based selection", loadBased.TranscodeNode)
+	}
+}
+
+func TestPrepareTransportV3LocalFallbackRejectsUnavailableTransformations(t *testing.T) {
+	handler := NewPlaybackHandler(playback.NewSessionManager(0, 0))
+	presetLocalRegistryV3(handler, playback.NewTransformationRegistryV3([]playback.TransformationSpecV3{
+		{Name: "video_to_h264", RecipeVersion: "1"},
+		{Name: "audio_to_aac", RecipeVersion: "1"},
+	}))
+	plan := &playback.PlanV3{
+		PlanID:   "plan:local-capability",
+		Delivery: playback.DeliveryTranscodeHLSV3,
+		Transformations: []playback.TransformationV3{
+			{Name: "video_to_h264", Executor: "server", RecipeVersion: "1"},
+			{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"},
+		},
+	}
+	request := httptest.NewRequest(http.MethodPost, "/", nil)
+	_, transportErr := handler.prepareTransportV3(request, &playback.Session{ID: "session-local-capability"}, v3HandlerFixtureFile(t), playback.PlannerResultV3{Plan: plan, PlayMethod: playback.PlayTranscode, TargetVideoCodec: "h264", TargetAudioCodec: "aac"})
+	if transportErr == nil || transportErr.reason != "transcode_node_capability_unavailable" || !transportErr.retryable {
+		t.Fatalf("transport error = %#v", transportErr)
+	}
+}
+
+func TestPlanRequiresServerTransformationsV3(t *testing.T) {
+	if planRequiresServerTransformationsV3(nil) {
+		t.Fatal("nil plan must not require server transformations")
+	}
+	clientOnly := &playback.PlanV3{Transformations: []playback.TransformationV3{{Name: playback.ClientDV7ToDV81V3, Executor: "client", RecipeVersion: "1"}}}
+	if planRequiresServerTransformationsV3(clientOnly) {
+		t.Fatal("client-executed transformations must not require a server executor")
+	}
+	server := &playback.PlanV3{Transformations: []playback.TransformationV3{{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"}}}
+	if !planRequiresServerTransformationsV3(server) {
+		t.Fatal("server-executed transformations must require executor validation")
+	}
+}
@@ -88,6 +88,17 @@ func NewPlanner(proxies *ProxyPool, transcodes *TranscodePool) *Planner {
 // bandwidth-cap admission. Re-planning the same session replaces its previous
 // reservation, so quality switches don't double-count.
 func (p *Planner) PlanSession(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int) Plan {
+	return p.PlanSessionWith(sessionID, currentTranscodeURL, needsTranscode, estBitrateKbps, nil)
+}
+
+// PlanSessionWith behaves like PlanSession but restricts transcode-node
+// selection to nodes accepted by eligible (nil accepts every node). Capability
+// -aware playback planning uses it so a recipe that only some pooled nodes
+// can execute is never load-balanced onto a node that cannot. The predicate
+// runs under the planner lock and must be cheap and non-blocking (a set
+// lookup, never a network call). Group health is still computed over the
+// full pool: eligibility narrows selection, not co-location semantics.
+func (p *Planner) PlanSessionWith(sessionID, currentTranscodeURL string, needsTranscode bool, estBitrateKbps int, eligible func(*Node) bool) Plan {
 	if p == nil {
 		return Plan{}
 	}
@@ -106,6 +117,15 @@ func (p *Planner) PlanSession(sessionID, currentTranscodeURL string, needsTransc
 	proxies := p.proxies.Nodes()
 	transcodes := p.transcodes.Nodes()
 	groupHealthy := groupHealth(proxies, transcodes)
+	if eligible != nil {
+		filtered := make([]*Node, 0, len(transcodes))
+		for _, node := range transcodes {
+			if eligible(node) {
+				filtered = append(filtered, node)
+			}
+		}
+		transcodes = filtered
+	}

 	var plan Plan
 	if needsTranscode {
@@ -131,6 +151,24 @@ func (p *Planner) PlanSession(sessionID, currentTranscodeURL string, needsTransc
 	return plan
 }

+// TranscodeNodeURLs lists the URLs of every enabled pooled transcode node,
+// healthy or not: capability planning wants the deployment's toolchain, and
+// an unreachable node excludes itself when its capability fetch fails. An
+// empty slice means no nodes are pooled.
+func (p *Planner) TranscodeNodeURLs() []string {
+	if p == nil || p.transcodes == nil {
+		return nil
+	}
+	nodes := p.transcodes.Nodes()
+	urls := make([]string, 0, len(nodes))
+	for _, node := range nodes {
+		if node != nil && node.URL != "" {
+			urls = append(urls, node.URL)
+		}
+	}
+	return urls
+}
+
 // ReleaseSession removes a provisional node reservation when playback setup
 // fails or falls back locally before a node health report can account for it.
 func (p *Planner) ReleaseSession(sessionID string) {
@@ -59,6 +59,30 @@ func TestPlanTranscodePairsProxyFromSameGroup(t *testing.T) {
 	}
 }

+func TestPlanSessionWithRestrictsEligibleTranscodeNodes(t *testing.T) {
+	f := newFixture(nil, []*Node{
+		transcodeNode(1, "http://tc-a", nil, 0),
+		transcodeNode(2, "http://tc-b", nil, 5),
+	})
+	eligible := func(n *Node) bool { return n != nil && n.URL == "http://tc-b" }
+
+	plan := f.planner.PlanSessionWith("s1", "", true, 0, eligible)
+	if plan.TranscodeNode == nil || plan.TranscodeNode.URL != "http://tc-b" {
+		t.Fatalf("expected the eligible node despite its higher load, got %+v", plan.TranscodeNode)
+	}
+	if none := f.planner.PlanSessionWith("s2", "", true, 0, func(*Node) bool { return false }); none.TranscodeNode != nil {
+		t.Fatalf("no eligible node must select nothing, got %+v", none.TranscodeNode)
+	}
+	// Soft affinity to the session's current node must not survive the
+	// current node becoming ineligible.
+	if sticky := f.planner.PlanSessionWith("s3", "http://tc-a", true, 0, eligible); sticky.TranscodeNode == nil || sticky.TranscodeNode.URL != "http://tc-b" {
+		t.Fatalf("affinity to an ineligible node must yield to an eligible one, got %+v", sticky.TranscodeNode)
+	}
+	if unrestricted := f.planner.PlanSessionWith("s4", "", true, 0, nil); unrestricted.TranscodeNode == nil || unrestricted.TranscodeNode.URL != "http://tc-a" {
+		t.Fatalf("nil predicate must behave like PlanSession, got %+v", unrestricted.TranscodeNode)
+	}
+}
+
 func TestReleaseSessionDropsProvisionalReservation(t *testing.T) {
 	node := transcodeNode(1, "http://tc-1", nil, 0)
 	node.MaxJobs = intPtr(1)
@@ -16,17 +16,43 @@ type PlannerSettingsV3 struct {
 }

 type PlannerInputV3 struct {
-	Request             StartRequestV3
-	RequestedFile       *models.MediaFile
-	EffectiveFile       *models.MediaFile
-	AudioTrackIndex     int
-	Settings            PlannerSettingsV3
-	Registry            *TransformationRegistryV3
+	Request         StartRequestV3
+	RequestedFile   *models.MediaFile
+	EffectiveFile   *models.MediaFile
+	AudioTrackIndex int
+	Settings        PlannerSettingsV3
+	// Registry holds the transformations the local binary can execute.
+	// Progressive remux routes always gate on it: they run in this process.
+	Registry *TransformationRegistryV3
+	// HLSRegistry optionally widens transformation availability for HLS
+	// deliveries, which can execute on pooled transcode nodes as well as
+	// locally. Nil means HLS routes gate on Registry alone. It is a lazy
+	// producer because building the widened registry can touch the network
+	// (node capability fetches): the planner only invokes it when a route
+	// decision genuinely depends on node capabilities, so direct-play and
+	// other source-preserving starts never pay for it. Producers must
+	// return a superset of Registry (local ∪ node capabilities) and should
+	// memoize; the transport layer re-validates whichever executor is
+	// actually selected.
+	HLSRegistry         func() *TransformationRegistryV3
 	Now                 time.Time
 	AttemptedKeys       []string
 	AdditionalSubtitles []SubtitleInventoryEntryV3
 }

+// hlsRegistry resolves the registry HLS deliveries gate on: the widened
+// local∪node registry when provided, otherwise the local one. Callers must
+// keep it behind short-circuits so transformation-free routes never force
+// the lazy producer to run.
+func (input PlannerInputV3) hlsRegistry() *TransformationRegistryV3 {
+	if input.HLSRegistry != nil {
+		if widened := input.HLSRegistry(); widened != nil {
+			return widened
+		}
+	}
+	return input.Registry
+}
+
 type PlannerResultV3 struct {
 	Plan                        *PlanV3
 	Terminal                    *TerminalV3
@@ -89,7 +115,17 @@ func PlanPlaybackV3(input PlannerInputV3) PlannerResultV3 {
 		audioClaims.Reason = "no_audio_track"
 	}
 	containerOK := containsFoldV3(input.Request.Capabilities.Containers, source.Container)
-	dvStripEligible := canStripDolbyVisionToHDR10V3(source, input.Request, input.Registry)
+	hlsEngineOK := engineAvailableV3(input.Request, EngineMedia3HLSV3)
+	// DV strip eligibility is split by executor pool: a progressive remux
+	// executes on this process's ffmpeg, while an HLS remux may run on a
+	// pooled transcode node advertising the transformation. Node capability
+	// only counts when the client can actually run an HLS delivery, and the
+	// widened registry is consulted lazily so non-DV sources never touch it.
+	dvStripEligibleLocal := canStripDolbyVisionToHDR10V3(source, input.Request, input.Registry)
+	dvStripEligible := dvStripEligibleLocal
+	if !dvStripEligible && hlsEngineOK && source.DynamicRange == "dolby_vision" {
+		dvStripEligible = canStripDolbyVisionToHDR10V3(source, input.Request, input.hlsRegistry())
+	}
 	clientDV81Eligible := canClientTransformDV7ToDV81V3(source, input.Request)
 	clientHDR10Eligible := canClientTransformDV7ToHDR10V3(source, input.Request)

@@ -221,8 +257,18 @@ func PlanPlaybackV3(input PlannerInputV3) PlannerResultV3 {
 		plan.Stream = StreamV3{Protocol: StreamHTTPProgressiveV3, Container: "mp4", MIMEType: "video/mp4", Headers: map[string]string{}, HeaderRefresh: HeaderRefreshSessionV3}
 		plan.DecisionReason = "container_normalization"
 		transcodeAudio := !audioOK
+		localAudioConvertOK := input.Registry != nil && input.Registry.Available("audio_to_aac")
 		if transcodeAudio {
-			if input.Registry == nil || !input.Registry.Available("audio_to_aac") {
+			// The HLS remux branch below can offload the conversion to a
+			// pooled node, but only for clients that can run an HLS
+			// delivery: a progressive-only client must keep this terminal
+			// (its retryable semantics included) rather than fall through
+			// to a generic adaptation_unavailable for a route it can never
+			// use. Short-circuit order keeps locally-capable planning from
+			// consulting node capabilities at all.
+			audioConvertOK := localAudioConvertOK ||
+				hlsEngineOK && input.hlsRegistry().Available("audio_to_aac")
+			if !audioConvertOK {
 				return terminalPlannerResultV3("audio_conversion_unsupported", "The required validated AAC conversion toolchain is unavailable.", true)
 			}
 			plan.EffectiveRecipe.AudioCodec = "aac"
@@ -243,7 +289,12 @@ func PlanPlaybackV3(input PlannerInputV3) PlannerResultV3 {
 		if !dvStrip {
 			applyCopiedVideoQuirksV3(&plan, source, input.Request, high10Quirk)
 		}
-		if remuxSubtitleOK {
+		// The progressive remux executes on this process's ffmpeg, so its
+		// server transformations must be locally available; when only pooled
+		// nodes carry them, the HLS remux below ships the same recipe on a
+		// node-offloadable delivery instead.
+		progressiveExecutable := (!transcodeAudio || localAudioConvertOK) && (!dvStrip || dvStripEligibleLocal)
+		if remuxSubtitleOK && progressiveExecutable {
 			plan.Subtitle = remuxSubtitle.Decision
 			plan.Claims.Subtitles = remuxSubtitle.Claims
 			finalizePlanIdentityV3(&plan, input.Request.PlaybackAttemptID)
@@ -259,7 +310,7 @@ func PlanPlaybackV3(input PlannerInputV3) PlannerResultV3 {
 			plan.Stream = StreamV3{Protocol: StreamHLSV3, Container: "hls", MIMEType: "application/vnd.apple.mpegurl", Headers: map[string]string{}, HeaderRefresh: HeaderRefreshSessionV3}
 			hlsTranscodeAudio := transcodeAudio
 			if audioQuirk, ok := hlsEAC3AudioCorrectionV3(source, input.Request); ok && !hlsTranscodeAudio {
-				if input.Registry == nil || !input.Registry.Available("audio_to_aac") {
+				if !input.hlsRegistry().Available("audio_to_aac") {
 					return terminalPlannerResultV3("audio_conversion_unsupported", "The device-specific HLS route requires the validated AAC conversion toolchain.", true)
 				}
 				hlsTranscodeAudio = true
@@ -320,7 +371,7 @@ func planVideoTranscodeV3(input PlannerInputV3, base PlanV3, source SourceDescri
 	if hdrTranscodeUnavailableV3(source) {
 		return terminalPlannerResultV3("hdr_transcode_unsupported", "This HDR source requires video encoding, but no validated HDR-preserving or tone-map recipe is installed.", false)
 	}
-	if input.Registry == nil || !input.Registry.Available("video_to_h264") || !input.Registry.Available("audio_to_aac") {
+	if !input.hlsRegistry().Available("video_to_h264") || !input.hlsRegistry().Available("audio_to_aac") {
 		return terminalPlannerResultV3("conversion_tool_unavailable", "The required validated H.264/AAC conversion toolchain is unavailable.", true)
 	}
 	plan := base
@@ -558,7 +609,7 @@ func videoTranscodeExecutableV3(input PlannerInputV3, source SourceDescriptorV3)
 	if hdrTranscodeUnavailableV3(source) {
 		return false
 	}
-	return input.Registry != nil && input.Registry.Available("video_to_h264") && input.Registry.Available("audio_to_aac")
+	return input.hlsRegistry().Available("video_to_h264") && input.hlsRegistry().Available("audio_to_aac")
 }

 func recipeFromSourceV3(source SourceDescriptorV3) EffectiveRecipeV3 {
@@ -0,0 +1,204 @@
+package playback
+
+import (
+	"testing"
+
+	"github.com/Silo-Server/silo-server/internal/models"
+)
+
+func staticHLSRegistryV3(registry *TransformationRegistryV3) func() *TransformationRegistryV3 {
+	return func() *TransformationRegistryV3 { return registry }
+}
+
+func TestTransformationRegistryWithAdvertised(t *testing.T) {
+	registry := NewTransformationRegistryV3([]TransformationSpecV3{
+		{Name: "audio_to_aac", RecipeVersion: "1"},
+		{Name: "video_to_h264", RecipeVersion: "1"},
+		{Name: "server_dv7_to_hdr10", RecipeVersion: "1", Available: true},
+	})
+	if got := registry.WithAdvertised(nil); got != registry {
+		t.Fatal("empty advertisement must return the receiver unchanged")
+	}
+	widened := registry.WithAdvertised([]TransformationV3{
+		{Name: "Audio_To_AAC", Executor: "server", RecipeVersion: "1"},
+		{Name: "video_to_h264", Executor: "server", RecipeVersion: "2"},
+		{Name: "made_up_transform", Executor: "server", RecipeVersion: "1"},
+	})
+	if !widened.Available("audio_to_aac") {
+		t.Fatal("a matching node advertisement must widen availability")
+	}
+	if widened.Available("video_to_h264") {
+		t.Fatal("a recipe-version mismatch must not widen availability")
+	}
+	if widened.Available("made_up_transform") {
+		t.Fatal("advertisements must not introduce specs the server does not define")
+	}
+	if !widened.Available("server_dv7_to_hdr10") {
+		t.Fatal("locally available specs must stay available")
+	}
+	if registry.Available("audio_to_aac") {
+		t.Fatal("widening must not mutate the receiver")
+	}
+	clientOnly := registry.WithAdvertised([]TransformationV3{{Name: "audio_to_aac", Executor: "client", RecipeVersion: "1"}})
+	if clientOnly.Available("audio_to_aac") {
+		t.Fatal("client-executor advertisements must not widen server availability")
+	}
+}
+
+// A deployment whose API host lacks the H.264/AAC toolchain must still plan
+// an HLS transcode when pooled transcode nodes advertise it, and must keep
+// the terminal when nothing does.
+func TestPlanPlaybackV3TranscodeOffloadsToNodeToolchain(t *testing.T) {
+	file := detailedFixtureFileV3()
+	file.VideoTracks[0].VideoRange = "SDR"
+	file.VideoTracks[0].VideoRangeType = "SDR"
+	file.VideoTracks[0].ColorTransfer = "bt709"
+	req := validStartRequestV3()
+	req.QualityPreference = "480p"
+	req.ClientFeatures = append(req.ClientFeatures, FeatureDetailedDecodeV3)
+	req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureDetailedDecodeV3)
+	req.Capabilities.VideoDecode = []VideoDecodeCapabilityV3{{Codec: "hevc", Profiles: []string{"main 10"}, Levels: []int{153}, BitDepths: []int{10}, MaxWidth: 3840, MaxHeight: 2160, MaxFrameRate: 60, MaxBitrateKbps: 80_000, Hardware: true}}
+	local := NewTransformationRegistryV3([]TransformationSpecV3{
+		{Name: "video_to_h264", RecipeVersion: "1"},
+		{Name: "audio_to_aac", RecipeVersion: "1"},
+	})
+	settings := PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true}
+
+	withoutNodes := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: local})
+	if withoutNodes.Terminal == nil || withoutNodes.Terminal.Reason != "conversion_tool_unavailable" {
+		t.Fatalf("without nodes = %s", ExplainPlannerResultV3(withoutNodes))
+	}
+
+	union := local.WithAdvertised([]TransformationV3{
+		{Name: "video_to_h264", Executor: "server", RecipeVersion: "1"},
+		{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"},
+	})
+	withNodes := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: local, HLSRegistry: staticHLSRegistryV3(union)})
+	if withNodes.Plan == nil || withNodes.Plan.Delivery != DeliveryTranscodeHLSV3 {
+		t.Fatalf("with nodes = %s", ExplainPlannerResultV3(withNodes))
+	}
+}
+
+// Audio conversion on the remux family must skip the locally-executed
+// progressive remux when only pooled nodes carry the AAC toolchain, shipping
+// the same recipe on the node-offloadable HLS remux delivery instead.
+func TestPlanPlaybackV3AudioAdaptationOffloadsToHLSRemux(t *testing.T) {
+	file := detailedFixtureFileV3()
+	file.AudioTracks[0] = models.AudioTrack{Codec: "truehd", Channels: 8, Layout: "7.1"}
+	file.CodecAudio = "truehd"
+	req := validStartRequestV3()
+	req.ClientFeatures = append(req.ClientFeatures, FeatureDetailedDecodeV3)
+	req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureDetailedDecodeV3)
+	req.Capabilities.VideoDecode = []VideoDecodeCapabilityV3{{Codec: "hevc", Profiles: []string{"main 10"}, Levels: []int{153}, BitDepths: []int{10}, MaxWidth: 3840, MaxHeight: 2160, MaxFrameRate: 60, MaxBitrateKbps: 80_000, Hardware: true}}
+	req.Capabilities.HDRDetails = &HDRCapabilitiesV3{HDR10: true}
+	local := NewTransformationRegistryV3([]TransformationSpecV3{{Name: "audio_to_aac", RecipeVersion: "1"}})
+	settings := PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true}
+
+	withoutNodes := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: local})
+	if withoutNodes.Terminal == nil || withoutNodes.Terminal.Reason != "audio_conversion_unsupported" {
+		t.Fatalf("without nodes = %s", ExplainPlannerResultV3(withoutNodes))
+	}
+
+	union := local.WithAdvertised([]TransformationV3{{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"}})
+	offloaded := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: local, HLSRegistry: staticHLSRegistryV3(union)})
+	if offloaded.Plan == nil || offloaded.Plan.Delivery != DeliveryRemuxHLSV3 || !offloaded.TranscodeAudio || offloaded.TargetAudioCodec != "aac" {
+		t.Fatalf("with nodes = %s", ExplainPlannerResultV3(offloaded))
+	}
+
+	// With the toolchain available locally the progressive remux keeps
+	// priority — offloadability must never demote a local-capable route.
+	localCapable := NewTransformationRegistryV3([]TransformationSpecV3{{Name: "audio_to_aac", RecipeVersion: "1", Available: true}})
+	preserved := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: localCapable, HLSRegistry: staticHLSRegistryV3(localCapable)})
+	if preserved.Plan == nil || preserved.Plan.Delivery != DeliveryRemuxProgressiveV3 {
+		t.Fatalf("local capable = %s", ExplainPlannerResultV3(preserved))
+	}
+}
+
+// A source that direct-plays must never trigger the lazy node-capability
+// producer: building the widened registry can touch the network, and dead
+// nodes must not add latency to starts that never use them.
+func TestPlanPlaybackV3DirectPlayNeverConsultsNodeCapabilities(t *testing.T) {
+	file := detailedFixtureFileV3()
+	req := validStartRequestV3()
+	req.ClientFeatures = append(req.ClientFeatures, FeatureDetailedDecodeV3)
+	req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureDetailedDecodeV3)
+	req.Capabilities.VideoDecode = []VideoDecodeCapabilityV3{{Codec: "hevc", Profiles: []string{"main 10"}, Levels: []int{153}, BitDepths: []int{10}, MaxWidth: 3840, MaxHeight: 2160, MaxFrameRate: 60, MaxBitrateKbps: 80_000, Hardware: true}}
+	req.Capabilities.HDRDetails = &HDRCapabilitiesV3{HDR10: true}
+
+	result := PlanPlaybackV3(PlannerInputV3{
+		Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0,
+		Settings: PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true},
+		Registry: testTransformationRegistryV3(),
+		HLSRegistry: func() *TransformationRegistryV3 {
+			t.Fatal("direct-play planning must not build the node capability registry")
+			return nil
+		},
+	})
+	if result.Plan == nil || result.Plan.Delivery != DeliveryOriginalHTTPV3 {
+		t.Fatalf("result = %s", ExplainPlannerResultV3(result))
+	}
+}
+
+// A progressive-only client (no HLS engine) cannot consume node-offloaded
+// conversions, so node capabilities must not suppress its specific retryable
+// audio terminal in favor of a generic non-retryable adaptation_unavailable.
+func TestPlanPlaybackV3NodeToolchainDoesNotMaskTerminalForProgressiveOnlyClient(t *testing.T) {
+	file := detailedFixtureFileV3()
+	file.AudioTracks[0] = models.AudioTrack{Codec: "truehd", Channels: 8, Layout: "7.1"}
+	file.CodecAudio = "truehd"
+	req := validStartRequestV3()
+	req.ClientFeatures = append(req.ClientFeatures, FeatureDetailedDecodeV3)
+	req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureDetailedDecodeV3)
+	req.Capabilities.VideoDecode = []VideoDecodeCapabilityV3{{Codec: "hevc", Profiles: []string{"main 10"}, Levels: []int{153}, BitDepths: []int{10}, MaxWidth: 3840, MaxHeight: 2160, MaxFrameRate: 60, MaxBitrateKbps: 80_000, Hardware: true}}
+	req.Capabilities.HDRDetails = &HDRCapabilitiesV3{HDR10: true}
+	delete(req.ClientPlaybackContext.Engines, string(EngineMedia3HLSV3))
+
+	local := NewTransformationRegistryV3([]TransformationSpecV3{{Name: "audio_to_aac", RecipeVersion: "1"}})
+	union := local.WithAdvertised([]TransformationV3{{Name: "audio_to_aac", Executor: "server", RecipeVersion: "1"}})
+	result := PlanPlaybackV3(PlannerInputV3{
+		Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0,
+		Settings: PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true},
+		Registry: local, HLSRegistry: staticHLSRegistryV3(union),
+	})
+	if result.Terminal == nil || result.Terminal.Reason != "audio_conversion_unsupported" || !result.Terminal.Retryable {
+		t.Fatalf("result = %s", ExplainPlannerResultV3(result))
+	}
+}
+
+// The Profile 7 HDR10 strip must ride the HLS remux when only pooled nodes
+// carry the dovi_rpu filter: the progressive remux executes locally and is
+// not eligible without the local filter.
+func TestPlanPlaybackV3Profile7StripOffloadsToHLSRemux(t *testing.T) {
+	file := detailedFixtureFileV3()
+	file.VideoTracks[0].DVProfile = 7
+	file.VideoTracks[0].DVBLCompatID = 6
+	file.VideoTracks[0].DVELPresent = false
+	file.VideoTracks[0].DVEnhancementLayer = ""
+	file.VideoTracks[0].VideoRange = "DolbyVision"
+	file.VideoTracks[0].VideoRangeType = "DOVIWithEL"
+	req := validStartRequestV3()
+	req.ClientFeatures = append(req.ClientFeatures, FeatureDetailedDecodeV3)
+	req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureDetailedDecodeV3)
+	req.Capabilities.VideoDecode = []VideoDecodeCapabilityV3{{Codec: "hevc", Profiles: []string{"main 10"}, Levels: []int{153}, BitDepths: []int{10}, MaxWidth: 3840, MaxHeight: 2160, MaxFrameRate: 60, MaxBitrateKbps: 80_000, Hardware: true}}
+	req.Capabilities.HDRDetails = &HDRCapabilitiesV3{HDR10: true, DolbyVisionProfiles: []int{5, 8}}
+	req.ClientPlaybackContext.Output.HDRDetails = req.Capabilities.HDRDetails
+	local := NewTransformationRegistryV3([]TransformationSpecV3{{Name: "server_dv7_to_hdr10", RecipeVersion: "1"}})
+	settings := PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true}
+
+	withoutNodes := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: local})
+	if withoutNodes.Terminal == nil || withoutNodes.Terminal.Reason != "hdr_transcode_unsupported" {
+		t.Fatalf("without nodes = %s", ExplainPlannerResultV3(withoutNodes))
+	}
+
+	union := local.WithAdvertised([]TransformationV3{{Name: "server_dv7_to_hdr10", Executor: "server", RecipeVersion: "1"}})
+	offloaded := PlanPlaybackV3(PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: settings, Registry: local, HLSRegistry: staticHLSRegistryV3(union)})
+	if offloaded.Plan == nil || offloaded.Plan.Delivery != DeliveryRemuxHLSV3 || offloaded.TargetVideoCodec != "copy" {
+		t.Fatalf("with nodes = %s", ExplainPlannerResultV3(offloaded))
+	}
+	if len(offloaded.Plan.Transformations) != 1 || offloaded.Plan.Transformations[0].Name != "server_dv7_to_hdr10" {
+		t.Fatalf("transformations = %#v", offloaded.Plan.Transformations)
+	}
+	if offloaded.Plan.EffectiveRecipe.DynamicRange != "hdr10" || !offloaded.Plan.Claims.Video.HDR10 {
+		t.Fatalf("claims = %#v", offloaded.Plan.Claims)
+	}
+}
@@ -5,6 +5,7 @@ import (
 	"context"
 	"os/exec"
 	"sort"
+	"strings"
 	"time"
 )

@@ -72,6 +73,39 @@ func (r *TransformationRegistryV3) Available(name string) bool {
 	return ok && spec.Available
 }

+// WithAdvertised returns a registry whose known specs are additionally marked
+// available when a pooled transcode node advertises the same server-executed
+// transformation at the same recipe version. Advertisements never introduce
+// new specs: the planner only selects transformations this server defines,
+// and pinning versions to the local spec guarantees a plan built from the
+// widened registry passes the per-node advertisement validation at transport
+// time. Returns the receiver unchanged when nothing new becomes available.
+func (r *TransformationRegistryV3) WithAdvertised(advertised []TransformationV3) *TransformationRegistryV3 {
+	if r == nil || len(advertised) == 0 {
+		return r
+	}
+	specs := make([]TransformationSpecV3, 0, len(r.entries))
+	changed := false
+	for _, spec := range r.entries {
+		if !spec.Available {
+			for _, remote := range advertised {
+				if strings.EqualFold(strings.TrimSpace(remote.Name), spec.Name) &&
+					strings.TrimSpace(remote.RecipeVersion) == spec.RecipeVersion &&
+					strings.EqualFold(strings.TrimSpace(remote.Executor), "server") {
+					spec.Available = true
+					changed = true
+					break
+				}
+			}
+		}
+		specs = append(specs, spec)
+	}
+	if !changed {
+		return r
+	}
+	return NewTransformationRegistryV3(specs)
+}
+
 func (r *TransformationRegistryV3) Advertised() []TransformationV3 {
 	if r == nil {
 		return nil