From b0a6aece39e1064e20248f50a9a5cf82b2bc8bd4 Mon Sep 17 00:00:00 2001
From: William Alvares <curruwilla@gmail.com>
Date: Mon, 13 Jul 2026 13:50:14 -0300
Subject: [PATCH 1/5] Add opt-in config gate for H.264 video pass-through

Adds an enable_video_passthrough deployment flag (default false). When
set, track composite and participant egress with HLS segment output no
longer force VideoDecoding/VideoEncoding at request time: the request
becomes a pass-through candidate if it subscribes to at most one video
track and carries no video encoding options (preset or advanced).

Any file, stream, or image output, and any input codec that does not
match the output codec at subscription time, falls back to the default
decode+encode pipeline with an informative log. With the flag off,
behavior is byte-for-byte identical to before.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md                           |   1 +
 pkg/config/base.go                  |   1 +
 pkg/config/output.go                |  24 ++-
 pkg/config/passthrough.go           |  97 ++++++++++++
 pkg/config/passthrough_test.go      | 234 ++++++++++++++++++++++++++++
 pkg/config/pipeline.go              |  13 ++
 pkg/pipeline/source/track_worker.go |   5 +
 7 files changed, 373 insertions(+), 2 deletions(-)
 create mode 100644 pkg/config/passthrough.go
 create mode 100644 pkg/config/passthrough_test.go

diff --git a/README.md b/README.md
index 381ac114..e71a97f7 100644
--- a/README.md
+++ b/README.md
@@ -69,6 +69,7 @@ logging:
 template_base: can be used to host custom templates (default http://localhost:<template_port>/)
 backup_storage: files will be moved here when uploads fail. location must have write access granted for all users
 enable_chrome_sandbox: if true, egress will run Chrome with sandboxing enabled. This requires a specific Docker setup, see below.
+enable_video_passthrough: if true, track composite and participant egress with HLS segment output may skip video decoding/encoding when the published track is already H.264 and no encoding options were requested. Falls back to transcoding otherwise (default false)
 cpu_cost: # optionally override cpu cost estimation, used when accepting or denying requests
   room_composite_cpu_cost: 3.0
   web_cpu_cost: 3.0
diff --git a/pkg/config/base.go b/pkg/config/base.go
index c3a93acc..363530e9 100644
--- a/pkg/config/base.go
+++ b/pkg/config/base.go
@@ -66,6 +66,7 @@ type BaseConfig struct {
 	LatencyOverrides              map[types.RequestType]LatencyConfig `yaml:"latency_overrides"`                  // latency overrides for different request types, experimental only, will be removed
 	EnableOneShotSenderReportSync bool                                `yaml:"enable_one_shot_sender_report_sync"` // temporary rollout flag enabling one-shot sender report correction for room composite / track requests that previously used audio PTS adjustment disabling
 	EnableSyncEngine              bool                                `yaml:"enable_sync_engine"`                 // use Chrome-inspired sync engine for improved cross-participant alignment and A/V sync
+	EnableVideoPassthrough        bool                                `yaml:"enable_video_passthrough"`           // allow skipping video decode/encode for single-track H.264 HLS egress when the input codec already matches the output
 	AudioTempoController          AudioTempoController                `yaml:"audio_tempo_controller"`             // audio tempo controller
 	TestOverrides                 TestOverrides                       `yaml:"test_overrides"`                     // set of config overrides for testing purposes
 }
diff --git a/pkg/config/output.go b/pkg/config/output.go
index 6c12ed04..8942f695 100644
--- a/pkg/config/output.go
+++ b/pkg/config/output.go
@@ -65,6 +65,7 @@ func (p *PipelineConfig) updateEncodedOutputs(req egress.EncodedOutput) error {
 		p.OutputCount.Inc()
 		p.FinalizationRequired = true
 		if p.VideoEnabled {
+			p.DisableVideoPassthrough("file output")
 			p.VideoEncoding = true
 		}
 
@@ -120,6 +121,7 @@ func (p *PipelineConfig) updateEncodedOutputs(req egress.EncodedOutput) error {
 		p.Outputs[types.EgressTypeStream] = []OutputConfig{conf}
 		p.OutputCount.Add(int32(len(stream.Urls)))
 		if p.VideoEnabled {
+			p.DisableVideoPassthrough("stream output")
 			p.VideoEncoding = true
 		}
 
@@ -161,7 +163,7 @@ func (p *PipelineConfig) updateEncodedOutputs(req egress.EncodedOutput) error {
 		p.Outputs[types.EgressTypeSegments] = []OutputConfig{conf}
 		p.OutputCount.Inc()
 		p.FinalizationRequired = true
-		if p.VideoEnabled {
+		if p.VideoEnabled && !p.VideoPassthrough {
 			p.VideoEncoding = true
 		}
 
@@ -189,6 +191,8 @@ func (p *PipelineConfig) updateEncodedOutputs(req egress.EncodedOutput) error {
 			return errors.ErrInvalidInput("audio_only images")
 		}
 
+		p.DisableVideoPassthrough("image output")
+
 		if len(p.Outputs) == 0 {
 			// enforce video only
 			p.AudioEnabled = false
@@ -214,6 +218,12 @@ func (p *PipelineConfig) updateEncodedOutputs(req egress.EncodedOutput) error {
 		return errors.ErrInvalidInput("output")
 	}
 
+	// if passthrough was disabled after the segment output was parsed (e.g. by
+	// an image output), the segment encoder still needs to be enabled
+	if p.VideoEnabled && !p.VideoPassthrough && p.Outputs[types.EgressTypeSegments] != nil {
+		p.VideoEncoding = true
+	}
+
 	return nil
 }
 
@@ -256,6 +266,7 @@ func (p *PipelineConfig) updateOutputs(req egress.EgressRequest) error {
 			p.OutputCount.Inc()
 			p.FinalizationRequired = true
 			if p.VideoEnabled {
+				p.DisableVideoPassthrough("file output")
 				p.VideoEncoding = true
 			}
 
@@ -319,6 +330,7 @@ func (p *PipelineConfig) updateOutputs(req egress.EgressRequest) error {
 			p.Outputs[egressType] = []OutputConfig{conf}
 			p.OutputCount.Add(int32(len(stream.Urls)))
 			if p.VideoEnabled {
+				p.DisableVideoPassthrough("stream output")
 				p.VideoEncoding = true
 			}
 
@@ -344,7 +356,7 @@ func (p *PipelineConfig) updateOutputs(req egress.EgressRequest) error {
 			p.Outputs[types.EgressTypeSegments] = []OutputConfig{conf}
 			p.OutputCount.Inc()
 			p.FinalizationRequired = true
-			if p.VideoEnabled {
+			if p.VideoEnabled && !p.VideoPassthrough {
 				p.VideoEncoding = true
 			}
 
@@ -355,6 +367,8 @@ func (p *PipelineConfig) updateOutputs(req egress.EgressRequest) error {
 				return errors.ErrInvalidInput("audio_only images")
 			}
 
+			p.DisableVideoPassthrough("image output")
+
 			conf, err := p.getImageConfig(o.Images, storage)
 			if err != nil {
 				return err
@@ -407,6 +421,12 @@ func (p *PipelineConfig) updateOutputs(req egress.EgressRequest) error {
 		p.KeyFrameInterval = StreamKeyframeInterval
 	}
 
+	// if passthrough was disabled after the segment output was parsed (outputs
+	// arrive in request order), the segment encoder still needs to be enabled
+	if p.VideoEnabled && !p.VideoPassthrough && hasSegments {
+		p.VideoEncoding = true
+	}
+
 	return nil
 }
 
diff --git a/pkg/config/passthrough.go b/pkg/config/passthrough.go
new file mode 100644
index 00000000..a373b221
--- /dev/null
+++ b/pkg/config/passthrough.go
@@ -0,0 +1,97 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+
+	"github.com/livekit/egress/pkg/types"
+)
+
+// Video pass-through skips the decode/encode stages and forwards the published
+// H.264 bitstream directly to the muxer. It is opt-in via the
+// enable_video_passthrough deployment flag and only ever downgrades to the
+// default transcoding pipeline, never the other way around.
+//
+// Eligibility is decided in two phases:
+//
+//  1. Request time (maybeEnableVideoPassthrough): the deployment flag is on,
+//     the request subscribes to at most one video track (TrackComposite or
+//     Participant - never composition), and no video encoding options were
+//     requested. Requiring the request to carry no explicit video options also
+//     guarantees no resolution/bitrate change is needed: the output follows
+//     whatever the track publishes.
+//  2. Output validation (disableVideoPassthrough callers in output.go) and
+//     subscription time (updatePreInitStateLocked in track_worker.go): any
+//     non-segment output, or an input codec that differs from the output
+//     codec, falls back to transcoding with an informative log.
+
+// maybeEnableVideoPassthrough marks the request as a pass-through candidate,
+// clearing the forced VideoDecoding flag. Must be called after encoding
+// options are applied and before outputs are parsed.
+func (p *PipelineConfig) maybeEnableVideoPassthrough(hasPreset bool, advanced *livekit.EncodingOptions) {
+	if !p.EnableVideoPassthrough || !p.VideoEnabled {
+		return
+	}
+
+	switch p.RequestType {
+	case types.RequestTypeTrackComposite, types.RequestTypeParticipant:
+		// single-publisher SDK sources, at most one video track subscribed
+	default:
+		logger.Infow("video passthrough not eligible",
+			"reason", "request type requires composition",
+			"requestType", p.RequestType)
+		return
+	}
+
+	if hasPreset {
+		logger.Infow("video passthrough not eligible",
+			"reason", "encoding preset requested")
+		return
+	}
+	if advanced != nil && advancedHasVideoOverrides(advanced) {
+		logger.Infow("video passthrough not eligible",
+			"reason", "advanced video encoding options requested")
+		return
+	}
+
+	p.VideoDecoding = false
+	p.VideoPassthrough = true
+	logger.Infow("video passthrough candidate",
+		"requestType", p.RequestType)
+}
+
+// DisableVideoPassthrough reverts a pass-through candidate to the default
+// decode path. Safe to call unconditionally.
+func (p *PipelineConfig) DisableVideoPassthrough(reason string) {
+	if !p.VideoPassthrough {
+		return
+	}
+	p.VideoPassthrough = false
+	p.VideoDecoding = true
+	logger.Infow("video passthrough disabled, falling back to transcoding",
+		"reason", reason)
+}
+
+func advancedHasVideoOverrides(advanced *livekit.EncodingOptions) bool {
+	return advanced.VideoCodec != livekit.VideoCodec_DEFAULT_VC ||
+		advanced.Width != 0 ||
+		advanced.Height != 0 ||
+		advanced.Depth != 0 ||
+		advanced.Framerate != 0 ||
+		advanced.VideoBitrate != 0 ||
+		advanced.KeyFrameInterval != 0
+}
diff --git a/pkg/config/passthrough_test.go b/pkg/config/passthrough_test.go
new file mode 100644
index 00000000..e55f508b
--- /dev/null
+++ b/pkg/config/passthrough_test.go
@@ -0,0 +1,234 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/logger"
+	"github.com/livekit/protocol/rpc"
+
+	"github.com/livekit/egress/pkg/types"
+)
+
+func newPassthroughTestConfig(enablePassthrough bool) *PipelineConfig {
+	return &PipelineConfig{
+		BaseConfig: BaseConfig{
+			Logging:                &logger.Config{Level: "info"},
+			TemplateBase:           "http://localhost:7980/",
+			EnableVideoPassthrough: enablePassthrough,
+		},
+		Outputs: make(map[types.EgressType][]OutputConfig),
+		Live:    true,
+	}
+}
+
+func segmentOutputs() []*livekit.SegmentedFileOutput {
+	return []*livekit.SegmentedFileOutput{{
+		FilenamePrefix: "test",
+		PlaylistName:   "playlist",
+	}}
+}
+
+func trackCompositeRequest(opts func(*livekit.TrackCompositeEgressRequest)) *rpc.StartEgressRequest {
+	tc := &livekit.TrackCompositeEgressRequest{
+		RoomName:       "room",
+		AudioTrackId:   "TR_audio",
+		VideoTrackId:   "TR_video",
+		SegmentOutputs: segmentOutputs(),
+	}
+	if opts != nil {
+		opts(tc)
+	}
+	return &rpc.StartEgressRequest{
+		EgressId: "EG_test",
+		Token:    "token",
+		WsUrl:    "wss://localhost:7880",
+		Request: &rpc.StartEgressRequest_TrackComposite{
+			TrackComposite: tc,
+		},
+	}
+}
+
+func TestVideoPassthroughEnabled(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	require.NoError(t, p.Update(trackCompositeRequest(nil)))
+
+	require.True(t, p.VideoPassthrough)
+	require.False(t, p.VideoDecoding)
+	require.False(t, p.VideoEncoding)
+	// HLS output still selects H264 as the target codec for the runtime check
+	require.Equal(t, types.MimeTypeH264, p.VideoOutCodec)
+}
+
+func TestVideoPassthroughParticipant(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := &rpc.StartEgressRequest{
+		EgressId: "EG_test",
+		Token:    "token",
+		WsUrl:    "wss://localhost:7880",
+		Request: &rpc.StartEgressRequest_Participant{
+			Participant: &livekit.ParticipantEgressRequest{
+				RoomName:       "room",
+				Identity:       "publisher",
+				SegmentOutputs: segmentOutputs(),
+			},
+		},
+	}
+	require.NoError(t, p.Update(req))
+
+	require.True(t, p.VideoPassthrough)
+	require.False(t, p.VideoDecoding)
+	require.False(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughFlagDisabled(t *testing.T) {
+	p := newPassthroughTestConfig(false)
+	require.NoError(t, p.Update(trackCompositeRequest(nil)))
+
+	// identical to pre-change behavior
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughRoomCompositeFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := &rpc.StartEgressRequest{
+		EgressId: "EG_test",
+		Token:    "token",
+		WsUrl:    "wss://localhost:7880",
+		Request: &rpc.StartEgressRequest_RoomComposite{
+			RoomComposite: &livekit.RoomCompositeEgressRequest{
+				RoomName:       "room",
+				SegmentOutputs: segmentOutputs(),
+			},
+		},
+	}
+	require.NoError(t, p.Update(req))
+
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughPresetFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := trackCompositeRequest(func(tc *livekit.TrackCompositeEgressRequest) {
+		tc.Options = &livekit.TrackCompositeEgressRequest_Preset{
+			Preset: livekit.EncodingOptionsPreset_H264_720P_30,
+		}
+	})
+	require.NoError(t, p.Update(req))
+
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughAdvancedVideoOptionsFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := trackCompositeRequest(func(tc *livekit.TrackCompositeEgressRequest) {
+		tc.Options = &livekit.TrackCompositeEgressRequest_Advanced{
+			Advanced: &livekit.EncodingOptions{
+				Width:  1920,
+				Height: 1080,
+			},
+		}
+	})
+	require.NoError(t, p.Update(req))
+
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughAudioOnlyAdvancedOptionsAllowed(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := trackCompositeRequest(func(tc *livekit.TrackCompositeEgressRequest) {
+		tc.Options = &livekit.TrackCompositeEgressRequest_Advanced{
+			Advanced: &livekit.EncodingOptions{
+				AudioBitrate: 96,
+			},
+		}
+	})
+	require.NoError(t, p.Update(req))
+
+	require.True(t, p.VideoPassthrough)
+	require.False(t, p.VideoDecoding)
+	require.False(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughFileOutputFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := trackCompositeRequest(func(tc *livekit.TrackCompositeEgressRequest) {
+		tc.SegmentOutputs = nil
+		tc.FileOutputs = []*livekit.EncodedFileOutput{{
+			Filepath: "test.mp4",
+		}}
+	})
+	require.NoError(t, p.Update(req))
+
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughMixedOutputsFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := trackCompositeRequest(func(tc *livekit.TrackCompositeEgressRequest) {
+		tc.StreamOutputs = []*livekit.StreamOutput{{
+			Protocol: livekit.StreamProtocol_RTMP,
+			Urls:     []string{"rtmp://localhost/live/stream"},
+		}}
+	})
+	require.NoError(t, p.Update(req))
+
+	// segments + stream: stream requires an encoder, no passthrough
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughImageOutputFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	req := trackCompositeRequest(func(tc *livekit.TrackCompositeEgressRequest) {
+		tc.ImageOutputs = []*livekit.ImageOutput{{
+			CaptureInterval: 5,
+			FilenamePrefix:  "image",
+		}}
+	})
+	require.NoError(t, p.Update(req))
+
+	// images are parsed after segments; encoder must still be re-enabled
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+	require.True(t, p.VideoEncoding)
+}
+
+func TestVideoPassthroughRuntimeCodecFallback(t *testing.T) {
+	p := newPassthroughTestConfig(true)
+	require.NoError(t, p.Update(trackCompositeRequest(nil)))
+	require.True(t, p.VideoPassthrough)
+
+	// equivalent to the subscription-time check in updatePreInitStateLocked
+	p.DisableVideoPassthrough("input codec video/vp8 does not match output codec video/h264")
+
+	require.False(t, p.VideoPassthrough)
+	require.True(t, p.VideoDecoding)
+}
diff --git a/pkg/config/pipeline.go b/pkg/config/pipeline.go
index 2903c987..c626dad0 100644
--- a/pkg/config/pipeline.go
+++ b/pkg/config/pipeline.go
@@ -137,6 +137,7 @@ type VideoConfig struct {
 	VideoEnabled     bool
 	VideoDecoding    bool
 	VideoEncoding    bool
+	VideoPassthrough bool
 	VideoOutCodec    types.MimeType
 	VideoProfile     types.Profile
 	Width            int32
@@ -347,16 +348,22 @@ func (p *PipelineConfig) Update(request *rpc.StartEgressRequest) error {
 		}
 
 		// encoding options
+		var hasPreset bool
+		var advanced *livekit.EncodingOptions
 		switch opts := req.Participant.Options.(type) {
 		case *livekit.ParticipantEgressRequest_Preset:
+			hasPreset = true
 			p.applyPreset(opts.Preset)
 
 		case *livekit.ParticipantEgressRequest_Advanced:
+			advanced = opts.Advanced
 			if err := p.applyAdvanced(opts.Advanced); err != nil {
 				return err
 			}
 		}
 
+		p.maybeEnableVideoPassthrough(hasPreset, advanced)
+
 		// output params
 		if err := p.updateEncodedOutputs(req.Participant); err != nil {
 			return err
@@ -388,16 +395,22 @@ func (p *PipelineConfig) Update(request *rpc.StartEgressRequest) error {
 		}
 
 		// encoding options
+		var hasPreset bool
+		var advanced *livekit.EncodingOptions
 		switch opts := req.TrackComposite.Options.(type) {
 		case *livekit.TrackCompositeEgressRequest_Preset:
+			hasPreset = true
 			p.applyPreset(opts.Preset)
 
 		case *livekit.TrackCompositeEgressRequest_Advanced:
+			advanced = opts.Advanced
 			if err := p.applyAdvanced(opts.Advanced); err != nil {
 				return err
 			}
 		}
 
+		p.maybeEnableVideoPassthrough(hasPreset, advanced)
+
 		// output params
 		if err := p.updateEncodedOutputs(req.TrackComposite); err != nil {
 			return err
diff --git a/pkg/pipeline/source/track_worker.go b/pkg/pipeline/source/track_worker.go
index 49e3aa02..da7a509a 100644
--- a/pkg/pipeline/source/track_worker.go
+++ b/pkg/pipeline/source/track_worker.go
@@ -226,10 +226,15 @@ func (s *SDKSource) updatePreInitStateLocked(op Operation, ts *config.TrackSourc
 			s.VideoOutCodec = ts.MimeType
 		}
 		if s.VideoInCodec != s.VideoOutCodec {
+			s.DisableVideoPassthrough(fmt.Sprintf("input codec %s does not match output codec %s", s.VideoInCodec, s.VideoOutCodec))
 			s.VideoDecoding = true
 			if len(s.GetEncodedOutputs()) > 0 {
 				s.VideoEncoding = true
 			}
+		} else if s.VideoPassthrough {
+			logger.Infow("video passthrough active",
+				"trackID", ts.TrackID,
+				"videoCodec", s.VideoInCodec)
 		}
 		s.VideoTrack = ts
 	}

From 16e0b9ff2c8a1c62c835a11e7fac4c0acb07719c Mon Sep 17 00:00:00 2001
From: William Alvares <curruwilla@gmail.com>
Date: Mon, 13 Jul 2026 13:50:29 -0300
Subject: [PATCH 2/5] Bridge splitmuxsink GstForceKeyUnit events to publisher
 PLI

In pass-through mode there is no local encoder, so the GstForceKeyUnit
events emitted by splitmuxsink (send-keyframe-requests=true) at segment
boundaries died unanswered. The keyframe probe now installs an upstream
event probe on the parser src pad when video decoding is disabled and
bridges those events to the appwriter's existing PLI path, reusing its
1s throttle rather than adding a second one.

Both the event receipt and the actual PLI send are logged with
nanosecond timestamps so the egress->SFU->publisher keyframe delay can
be measured in staging.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 pkg/pipeline/builder/keyframe_probe.go      | 49 +++++++++++++++++++--
 pkg/pipeline/builder/keyframe_probe_test.go | 21 +++++++++
 pkg/pipeline/builder/video.go               |  5 ++-
 pkg/pipeline/source/sdk/appwriter.go        |  7 ++-
 4 files changed, 77 insertions(+), 5 deletions(-)

diff --git a/pkg/pipeline/builder/keyframe_probe.go b/pkg/pipeline/builder/keyframe_probe.go
index e33b5d29..8fbb211f 100644
--- a/pkg/pipeline/builder/keyframe_probe.go
+++ b/pkg/pipeline/builder/keyframe_probe.go
@@ -43,12 +43,19 @@ const (
 //
 // After that it's mostly inert: it just patches the occasional baseparse
 // missing-PTS buffer from the last good value, so downstream stays monotonic.
+//
+// In video pass-through mode there is no local encoder, so upstream
+// GstForceKeyUnit events (sent by splitmuxsink at segment boundaries) would
+// otherwise die unanswered. When forwardForceKeyUnit is set, the probe bridges
+// them to a PLI to the publisher, reusing the same onRequestPLI callback (and
+// therefore the appwriter's existing 1s PLI throttle).
 type keyframeProbe struct {
 	trackID  string
 	mimeType types.MimeType
 
-	srcPad     *gst.Pad
-	srcProbeID uint64
+	srcPad       *gst.Pad
+	srcProbeID   uint64
+	eventProbeID uint64
 
 	onRequestPLI func()
 
@@ -67,7 +74,7 @@ type keyframeProbe struct {
 	totalIntervals   int
 }
 
-func newKeyframeProbe(trackID string, mimeType types.MimeType, element *gst.Element, onRequestPLI func()) (*keyframeProbe, error) {
+func newKeyframeProbe(trackID string, mimeType types.MimeType, element *gst.Element, onRequestPLI func(), forwardForceKeyUnit bool) (*keyframeProbe, error) {
 	srcPad := element.GetStaticPad("src")
 	if srcPad == nil {
 		return nil, errors.ErrGstPipelineError(newMissingPadError(element.GetName(), "src"))
@@ -84,6 +91,9 @@ func newKeyframeProbe(trackID string, mimeType types.MimeType, element *gst.Elem
 	p.keyframePending = true
 
 	p.srcProbeID = srcPad.AddProbe(gst.PadProbeTypeBuffer, p.onSrcBuffer)
+	if forwardForceKeyUnit {
+		p.eventProbeID = srcPad.AddProbe(gst.PadProbeTypeEventUpstream, p.onUpstreamEvent)
+	}
 
 	return p, nil
 }
@@ -93,10 +103,43 @@ func (p *keyframeProbe) Close() {
 
 	if p.srcPad != nil {
 		p.srcPad.RemoveProbe(p.srcProbeID)
+		if p.eventProbeID != 0 {
+			p.srcPad.RemoveProbe(p.eventProbeID)
+		}
 		p.srcPad = nil
 	}
 }
 
+// onUpstreamEvent bridges splitmuxsink's GstForceKeyUnit requests to a PLI to
+// the publisher. onRequestPLI routes through the appwriter, whose existing 1s
+// throttle applies - no additional throttling here. Both the event receipt
+// (here) and the actual PLI send (appwriter) are logged with timestamps so the
+// egress->SFU->publisher keyframe delay can be measured.
+func (p *keyframeProbe) onUpstreamEvent(_ *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn {
+	return p.processUpstreamEvent(info.GetEvent())
+}
+
+func (p *keyframeProbe) processUpstreamEvent(event *gst.Event) gst.PadProbeReturn {
+	if event == nil || event.Type() != gst.EventTypeCustomUpstream {
+		return gst.PadProbeOK
+	}
+	s := event.GetStructure()
+	if s == nil || s.Name() != "GstForceKeyUnit" {
+		return gst.PadProbeOK
+	}
+
+	fields := []any{"receivedAt", time.Now().UnixNano()}
+	if runningTime, err := s.GetValue("running-time"); err == nil {
+		fields = append(fields, "runningTime", runningTime)
+	}
+	p.logger.Infow("force key unit event received, requesting PLI", fields...)
+
+	if p.onRequestPLI != nil {
+		p.onRequestPLI()
+	}
+	return gst.PadProbeOK
+}
+
 func (p *keyframeProbe) onSrcBuffer(_ *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn {
 	buffer := info.GetBuffer()
 	if buffer == nil {
diff --git a/pkg/pipeline/builder/keyframe_probe_test.go b/pkg/pipeline/builder/keyframe_probe_test.go
index 2333832d..d1dd4f4d 100644
--- a/pkg/pipeline/builder/keyframe_probe_test.go
+++ b/pkg/pipeline/builder/keyframe_probe_test.go
@@ -129,6 +129,27 @@ func TestKeyframeProbe_MissingPTSWithoutPriorValidDrops(t *testing.T) {
 	require.Equal(t, int32(0), pliCount.Load())
 }
 
+func TestKeyframeProbe_ForceKeyUnitEventRequestsPLI(t *testing.T) {
+	initGStreamer(t)
+
+	var pliCount atomic.Int32
+	p := newTestProbe(types.MimeTypeH264, func() { pliCount.Add(1) })
+
+	// splitmuxsink keyframe request must be bridged to a PLI and forwarded
+	ev := gst.NewCustomEvent(gst.EventTypeCustomUpstream, gst.NewStructure("GstForceKeyUnit"))
+	require.Equal(t, gst.PadProbeOK, p.processUpstreamEvent(ev))
+	require.Equal(t, int32(1), pliCount.Load())
+
+	// other custom upstream events are ignored
+	other := gst.NewCustomEvent(gst.EventTypeCustomUpstream, gst.NewStructure("SomethingElse"))
+	require.Equal(t, gst.PadProbeOK, p.processUpstreamEvent(other))
+	require.Equal(t, int32(1), pliCount.Load())
+
+	// nil events are ignored
+	require.Equal(t, gst.PadProbeOK, p.processUpstreamEvent(nil))
+	require.Equal(t, int32(1), pliCount.Load())
+}
+
 func TestKeyframeProbe_MissingPTSPostKeyframeRestoresPTSAndForwards(t *testing.T) {
 	initGStreamer(t)
 
diff --git a/pkg/pipeline/builder/video.go b/pkg/pipeline/builder/video.go
index 674e9d5c..6e37b0d5 100644
--- a/pkg/pipeline/builder/video.go
+++ b/pkg/pipeline/builder/video.go
@@ -387,7 +387,10 @@ func (b *VideoBin) addAppSrcBin(ts *config.TrackSource) error {
 }
 
 func (b *VideoBin) attachKeyframeProbe(ts *config.TrackSource, name string, element *gst.Element) error {
-	probe, err := newKeyframeProbe(ts.TrackID, ts.MimeType, element, ts.OnKeyframeRequired)
+	// without a local encoder, splitmuxsink keyframe requests must be bridged
+	// to a PLI to the publisher
+	forwardForceKeyUnit := !b.conf.VideoDecoding
+	probe, err := newKeyframeProbe(ts.TrackID, ts.MimeType, element, ts.OnKeyframeRequired, forwardForceKeyUnit)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/pipeline/source/sdk/appwriter.go b/pkg/pipeline/source/sdk/appwriter.go
index 31e608f7..9c58b578 100644
--- a/pkg/pipeline/source/sdk/appwriter.go
+++ b/pkg/pipeline/source/sdk/appwriter.go
@@ -220,7 +220,12 @@ func NewAppWriter(
 	w.sendPLI = func() {}
 	if track.Kind() == webrtc.RTPCodecTypeVideo {
 		w.pliThrottle = core.NewThrottle(time.Second)
-		w.sendPLI = func() { w.pliThrottle(func() { rp.WritePLI(track.SSRC()) }) }
+		w.sendPLI = func() {
+			w.pliThrottle(func() {
+				w.logger.Infow("sending PLI to publisher", "sentAt", time.Now().UnixNano(), "ssrc", track.SSRC())
+				rp.WritePLI(track.SSRC())
+			})
+		}
 		opts = append(opts, jitter.WithPacketLossHandler(func(uint64, uint64) { w.sendPLI() }))
 	}
 

From 8831251107ef5561b84b4d1bd545853d2c5b5db9 Mon Sep 17 00:00:00 2001
From: William Alvares <curruwilla@gmail.com>
Date: Mon, 13 Jul 2026 14:24:04 -0300
Subject: [PATCH 3/5] Add freeze fallback for mute/disconnect in video
 pass-through

Without a decoder, the input-selector + videotestsrc fallback does not
exist in pass-through mode: a muted or disconnected publisher stops the
buffer flow and splitmuxsink stalls, freezing the HLS playlist and the
viewer's player.

The selector machinery is now also built in pass-through mode, operating
on the encoded H.264 stream, with the videotestsrc replaced by a freeze
injector: an auxiliary appsrc that caches the last keyframe seen on the
real track and re-pushes it (2fps, do-timestamp PTS) while the fallback
pad is selected. Since every injected frame is an IDR, splitmuxsink
keeps cutting segments at the nominal duration while frozen.

Switching uses the existing mute/unmute/removed callbacks - the
appwriter's inactivity path (jitter buffer latency) already acts as the
sample watchdog, and unmute re-selects the track pad, dropping delta
frames until the PLI-requested keyframe arrives. A secondary stall
detector (2x segment duration) logs if data stops without any mute
event firing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 pkg/pipeline/builder/freeze_injector.go      | 195 +++++++++++++++++++
 pkg/pipeline/builder/freeze_injector_test.go |  98 ++++++++++
 pkg/pipeline/builder/keyframe_probe.go       |   8 +
 pkg/pipeline/builder/video.go                | 115 +++++++++--
 4 files changed, 401 insertions(+), 15 deletions(-)
 create mode 100644 pkg/pipeline/builder/freeze_injector.go
 create mode 100644 pkg/pipeline/builder/freeze_injector_test.go

diff --git a/pkg/pipeline/builder/freeze_injector.go b/pkg/pipeline/builder/freeze_injector.go
new file mode 100644
index 00000000..4275d017
--- /dev/null
+++ b/pkg/pipeline/builder/freeze_injector.go
@@ -0,0 +1,195 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package builder
+
+import (
+	"io"
+	"time"
+
+	"github.com/frostbyte73/core"
+	"github.com/go-gst/go-gst/gst"
+	"github.com/go-gst/go-gst/gst/app"
+	"github.com/linkdata/deadlock"
+	"go.uber.org/atomic"
+
+	"github.com/livekit/protocol/logger"
+)
+
+const freezeFrameInterval = 500 * time.Millisecond
+
+// freezeInjector is the pass-through equivalent of the decoded path's
+// videotestsrc fallback. Without a local encoder there is no way to generate
+// filler video, so instead it caches the last H.264 keyframe seen on the real
+// track and, while the fallback selector pad is active (publisher muted,
+// disconnected, or stalled), re-pushes that keyframe on an auxiliary appsrc so
+// splitmuxsink keeps cutting segments and the HLS playlist keeps advancing.
+//
+// Repeated IDRs form a valid H.264 stream, and since every injected frame is a
+// keyframe, splitmuxsink splits exactly at the configured segment duration
+// while frozen. PTS comes from the appsrc's do-timestamp (pipeline running
+// time), which keeps it monotonic with the real track's synchronizer-driven
+// PTS; the selector pad probes drop any small overlap on switchover.
+type freezeInjector struct {
+	src            *app.Source
+	fallbackActive func() bool
+	stallAfter     time.Duration
+
+	mu      deadlock.Mutex
+	frame   []byte
+	capsSet bool
+
+	lastRealBuffer atomic.Int64 // UnixNano of the last real track buffer
+	frozen         atomic.Bool
+	framesInjected atomic.Uint64
+	lastStallWarn  atomic.Int64
+
+	started core.Fuse
+	stopped core.Fuse
+
+	logger logger.Logger
+}
+
+func newFreezeInjector(src *app.Source, stallAfter time.Duration, fallbackActive func() bool) *freezeInjector {
+	return &freezeInjector{
+		src:            src,
+		fallbackActive: fallbackActive,
+		stallAfter:     stallAfter,
+		logger:         logger.GetLogger().WithValues("component", "freeze_injector"),
+	}
+}
+
+// OnRealBuffer must be called for every buffer flowing on the real track's
+// parser pad. It feeds the stall detector.
+func (f *freezeInjector) OnRealBuffer() {
+	f.lastRealBuffer.Store(time.Now().UnixNano())
+}
+
+// CacheKeyframe stores a copy of the latest keyframe and the current stream
+// caps. The injection loop starts on the first cached keyframe - before that
+// there is nothing valid to inject.
+func (f *freezeInjector) CacheKeyframe(buffer *gst.Buffer, pad *gst.Pad) {
+	mapInfo := buffer.Map(gst.MapRead)
+	if mapInfo == nil {
+		return
+	}
+	data, err := io.ReadAll(mapInfo.Reader())
+	buffer.Unmap()
+	if err != nil {
+		return
+	}
+
+	f.cache(data, pad.GetCurrentCaps())
+}
+
+func (f *freezeInjector) cache(data []byte, caps *gst.Caps) {
+	if len(data) == 0 {
+		return
+	}
+
+	f.mu.Lock()
+	f.frame = data
+	if !f.capsSet && caps != nil {
+		f.src.SetCaps(caps)
+		f.capsSet = true
+	}
+	f.mu.Unlock()
+
+	f.started.Once(func() {
+		go f.run()
+	})
+}
+
+// Stop ends the injection loop. When sendEOS is set, the auxiliary appsrc
+// emits EOS so pipeline shutdown can propagate through the selector when the
+// fallback pad is the active one.
+func (f *freezeInjector) Stop(sendEOS bool) {
+	f.stopped.Once(func() {
+		if sendEOS {
+			if flow := f.src.EndStream(); flow != gst.FlowOK && flow != gst.FlowFlushing {
+				f.logger.Warnw("unexpected flow return on freeze src EOS", nil, "flowReturn", flow.String())
+			}
+		}
+	})
+}
+
+func (f *freezeInjector) run() {
+	ticker := time.NewTicker(freezeFrameInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-f.stopped.Watch():
+			return
+		case <-ticker.C:
+			if f.fallbackActive() {
+				f.injectFrame()
+			} else {
+				f.checkStall()
+			}
+		}
+	}
+}
+
+func (f *freezeInjector) injectFrame() {
+	f.mu.Lock()
+	data := f.frame
+	capsSet := f.capsSet
+	f.mu.Unlock()
+	if data == nil || !capsSet {
+		return
+	}
+
+	if f.frozen.CompareAndSwap(false, true) {
+		f.logger.Infow("video freeze started, repeating last keyframe",
+			"startedAt", time.Now().UnixNano())
+	}
+
+	buffer := gst.NewBufferFromBytes(data)
+	buffer.SetDuration(gst.ClockTime(uint64(freezeFrameInterval)))
+	if flow := f.src.PushBuffer(buffer); flow != gst.FlowOK {
+		// flushing while the pipeline spins up or shuts down is expected
+		if flow != gst.FlowFlushing {
+			f.logger.Warnw("freeze frame push failed", nil, "flowReturn", flow.String())
+		}
+		return
+	}
+	f.framesInjected.Inc()
+}
+
+func (f *freezeInjector) checkStall() {
+	if f.frozen.CompareAndSwap(true, false) {
+		f.logger.Infow("video freeze ended, track resumed",
+			"endedAt", time.Now().UnixNano(),
+			"framesInjected", f.framesInjected.Load())
+	}
+
+	lastReal := f.lastRealBuffer.Load()
+	if lastReal == 0 {
+		return
+	}
+	stalledFor := time.Since(time.Unix(0, lastReal))
+	if stalledFor <= f.stallAfter {
+		return
+	}
+
+	// The track pad is still selected but nothing is flowing and no
+	// mute/unsubscribe event fired. The appwriter's inactivity path should
+	// have switched to the fallback already - log so this shows up.
+	lastWarn := f.lastStallWarn.Load()
+	if time.Since(time.Unix(0, lastWarn)) > f.stallAfter && f.lastStallWarn.CompareAndSwap(lastWarn, time.Now().UnixNano()) {
+		f.logger.Warnw("video track stalled without mute event", nil,
+			"stalledFor", stalledFor)
+	}
+}
diff --git a/pkg/pipeline/builder/freeze_injector_test.go b/pkg/pipeline/builder/freeze_injector_test.go
new file mode 100644
index 00000000..e27b6615
--- /dev/null
+++ b/pkg/pipeline/builder/freeze_injector_test.go
@@ -0,0 +1,98 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package builder
+
+import (
+	"testing"
+	"time"
+
+	"github.com/go-gst/go-gst/gst"
+	"github.com/go-gst/go-gst/gst/app"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+func newTestInjector(t *testing.T, fallbackActive func() bool) *freezeInjector {
+	t.Helper()
+
+	element, err := gst.NewElementWithName("appsrc", "test_freeze_src")
+	require.NoError(t, err)
+	src := app.SrcFromElement(element)
+
+	return newFreezeInjector(src, 8*time.Second, fallbackActive)
+}
+
+func TestFreezeInjector_CacheStartsLoopOnce(t *testing.T) {
+	initGStreamer(t)
+
+	active := atomic.NewBool(false)
+	f := newTestInjector(t, active.Load)
+	defer f.Stop(false)
+
+	require.False(t, f.started.IsBroken())
+
+	caps := gst.NewCapsFromString("video/x-h264,stream-format=byte-stream,alignment=au")
+	f.cache([]byte{0, 0, 0, 1, 0x65, 1, 2, 3}, caps)
+
+	require.True(t, f.started.IsBroken(), "first cached keyframe should start the loop")
+	require.True(t, f.capsSet)
+	require.Equal(t, []byte{0, 0, 0, 1, 0x65, 1, 2, 3}, f.frame)
+
+	// later keyframes replace the cached frame without resetting caps
+	f.cache([]byte{0, 0, 0, 1, 0x65, 9}, nil)
+	require.Equal(t, []byte{0, 0, 0, 1, 0x65, 9}, f.frame)
+}
+
+func TestFreezeInjector_EmptyOrMissingDataIgnored(t *testing.T) {
+	initGStreamer(t)
+
+	f := newTestInjector(t, func() bool { return false })
+	defer f.Stop(false)
+
+	f.cache(nil, nil)
+	require.False(t, f.started.IsBroken())
+	require.Nil(t, f.frame)
+}
+
+func TestFreezeInjector_InjectRequiresFrameAndCaps(t *testing.T) {
+	initGStreamer(t)
+
+	f := newTestInjector(t, func() bool { return true })
+	defer f.Stop(false)
+
+	// no frame cached: no-op
+	f.injectFrame()
+	require.Equal(t, uint64(0), f.framesInjected.Load())
+	require.False(t, f.frozen.Load())
+
+	// frame cached: injection proceeds (appsrc queues internally until playing)
+	caps := gst.NewCapsFromString("video/x-h264,stream-format=byte-stream,alignment=au")
+	f.cache([]byte{0, 0, 0, 1, 0x65, 1}, caps)
+	f.injectFrame()
+	require.True(t, f.frozen.Load(), "freeze mode entered on first injection attempt")
+	require.Equal(t, uint64(1), f.framesInjected.Load())
+}
+
+func TestFreezeInjector_StallDetectorResetsFreezeState(t *testing.T) {
+	initGStreamer(t)
+
+	f := newTestInjector(t, func() bool { return false })
+	defer f.Stop(false)
+
+	f.frozen.Store(true)
+	f.OnRealBuffer()
+	f.checkStall()
+	require.False(t, f.frozen.Load(), "resumed track should clear freeze state")
+}
diff --git a/pkg/pipeline/builder/keyframe_probe.go b/pkg/pipeline/builder/keyframe_probe.go
index 8fbb211f..4c93f6b3 100644
--- a/pkg/pipeline/builder/keyframe_probe.go
+++ b/pkg/pipeline/builder/keyframe_probe.go
@@ -58,6 +58,7 @@ type keyframeProbe struct {
 	eventProbeID uint64
 
 	onRequestPLI func()
+	injector     *freezeInjector // pass-through only, nil otherwise
 
 	logger logger.Logger
 
@@ -168,6 +169,13 @@ func (p *keyframeProbe) processBuffer(buffer *gst.Buffer) gst.PadProbeReturn {
 
 	isKeyframe := buffer.GetFlags()&gst.BufferFlagDeltaUnit == 0
 
+	if p.injector != nil {
+		p.injector.OnRealBuffer()
+		if isKeyframe {
+			p.injector.CacheKeyframe(buffer, p.srcPad)
+		}
+	}
+
 	if isKeyframe {
 		if p.keyframePending {
 			p.keyframePending = false
diff --git a/pkg/pipeline/builder/video.go b/pkg/pipeline/builder/video.go
index 6e37b0d5..24fc1c1d 100644
--- a/pkg/pipeline/builder/video.go
+++ b/pkg/pipeline/builder/video.go
@@ -47,11 +47,20 @@ type VideoBin struct {
 	names       map[string]string
 	selector    *gst.Element
 	rawVideoTee *gst.Element
+	freeze      *freezeInjector
 
 	probesMu deadlock.Mutex
 	probes   map[string]*keyframeProbe
 }
 
+// useSelector reports whether the input-selector fallback machinery is built:
+// always in the decoded path (videotestsrc fallback), and in pass-through mode
+// (freeze injector fallback). Track egress without pass-through keeps the bare
+// appsrc chain.
+func (b *VideoBin) useSelector() bool {
+	return b.conf.VideoDecoding || b.conf.VideoPassthrough
+}
+
 // buildVideoQueue creates a queue for the video pipeline. For live sources the
 // queue is leaky (drops old buffers when full) to handle real-time overrun. For
 // non-live replay the queue is blocking so backpressure throttles the source.
@@ -225,7 +234,7 @@ func (b *VideoBin) resetVideoAppSrcBin(ts *config.TrackSource) error {
 	}
 
 	// If the stuck bin is the currently selected pad, switch to test src first
-	if b.conf.VideoDecoding && b.selectedPad == oldName {
+	if b.useSelector() && b.selectedPad == oldName {
 		if err := b.setSelectorPadLocked(videoTestSrcName); err != nil {
 			return err
 		}
@@ -256,7 +265,7 @@ func (b *VideoBin) resetVideoAppSrcBin(ts *config.TrackSource) error {
 		return fmt.Errorf("failed to build new video source bin: %w", err)
 	}
 
-	if b.conf.VideoDecoding {
+	if b.useSelector() {
 		b.createSrcPadLocked(ts.TrackID, name)
 	}
 
@@ -264,7 +273,7 @@ func (b *VideoBin) resetVideoAppSrcBin(ts *config.TrackSource) error {
 		return fmt.Errorf("failed to add new video source bin: %w", err)
 	}
 
-	if b.conf.VideoDecoding {
+	if b.useSelector() {
 		if err := b.setSelectorPadLocked(name); err != nil {
 			return err
 		}
@@ -330,11 +339,22 @@ func (b *VideoBin) buildSDKInput() error {
 	b.pads = make(map[string]*gst.Pad)
 	b.names = make(map[string]string)
 
-	// add selector first so pads can be created
-	if b.conf.VideoDecoding {
+	// add selector and fallback source first so pads can be created
+	if b.useSelector() {
 		if err := b.addSelector(); err != nil {
 			return err
 		}
+		b.bin.SetGetSrcPad(b.getSrcPad)
+
+		if b.conf.VideoDecoding {
+			if err := b.addVideoTestSrcBin(); err != nil {
+				return err
+			}
+		} else {
+			if err := b.addFreezeSrcBin(); err != nil {
+				return err
+			}
+		}
 	}
 
 	if b.conf.VideoTrack != nil {
@@ -343,19 +363,16 @@ func (b *VideoBin) buildSDKInput() error {
 		}
 	}
 
-	if b.conf.VideoDecoding {
-		b.bin.SetGetSrcPad(b.getSrcPad)
-
-		if err := b.addVideoTestSrcBin(); err != nil {
-			return err
-		}
+	if b.useSelector() {
 		if b.conf.VideoTrack == nil {
 			if err := b.setSelectorPad(videoTestSrcName); err != nil {
 				return err
 			}
 		}
-		if err := b.addDecodedVideoSink(); err != nil {
-			return err
+		if b.conf.VideoDecoding {
+			if err := b.addDecodedVideoSink(); err != nil {
+				return err
+			}
 		}
 	}
 
@@ -371,7 +388,7 @@ func (b *VideoBin) addAppSrcBin(ts *config.TrackSource) error {
 		return err
 	}
 
-	if b.conf.VideoDecoding {
+	if b.useSelector() {
 		b.createSrcPad(ts.TrackID, name)
 	}
 
@@ -379,7 +396,7 @@ func (b *VideoBin) addAppSrcBin(ts *config.TrackSource) error {
 		return err
 	}
 
-	if b.conf.VideoDecoding {
+	if b.useSelector() {
 		return b.setSelectorPad(name)
 	}
 
@@ -394,6 +411,8 @@ func (b *VideoBin) attachKeyframeProbe(ts *config.TrackSource, name string, elem
 	if err != nil {
 		return err
 	}
+	// feed the freeze injector's keyframe cache and stall detector
+	probe.injector = b.freeze
 	b.probesMu.Lock()
 	b.probes[name] = probe
 	b.probesMu.Unlock()
@@ -620,6 +639,16 @@ func (b *VideoBin) addSelector() error {
 		return errors.ErrGstPipelineError(err)
 	}
 
+	// pass-through carries encoded H.264 through the selector - no rate or raw
+	// caps adjustment possible without decoding
+	if !b.conf.VideoDecoding {
+		if err = b.bin.AddElements(inputSelector); err != nil {
+			return err
+		}
+		b.selector = inputSelector
+		return nil
+	}
+
 	videoRate, err := gst.NewElement("videorate")
 	if err != nil {
 		return errors.ErrGstPipelineError(err)
@@ -641,6 +670,62 @@ func (b *VideoBin) addSelector() error {
 	return nil
 }
 
+// addFreezeSrcBin is the pass-through counterpart of addVideoTestSrcBin: an
+// auxiliary appsrc that repeats the last cached keyframe while the fallback
+// pad is active (publisher muted or disconnected), so the HLS playlist keeps
+// advancing instead of stalling the player. It registers under the same
+// fallback pad name so all selector switching logic is shared with the
+// decoded path.
+func (b *VideoBin) addFreezeSrcBin() error {
+	freezeBin := b.bin.NewBin(videoTestSrcName)
+	if err := b.bin.AddSourceBin(freezeBin); err != nil {
+		return err
+	}
+
+	element, err := gst.NewElementWithName("appsrc", "video_freeze_src")
+	if err != nil {
+		return errors.ErrGstPipelineError(err)
+	}
+	freezeSrc := app.SrcFromElement(element)
+	freezeSrc.SetArg("format", "time")
+	if err = freezeSrc.SetProperty("is-live", true); err != nil {
+		return errors.ErrGstPipelineError(err)
+	}
+	// stamp injected frames with pipeline running time, keeping PTS monotonic
+	// with the real track's synchronizer-driven PTS
+	if err = freezeSrc.SetProperty("do-timestamp", true); err != nil {
+		return errors.ErrGstPipelineError(err)
+	}
+
+	queue, err := b.buildVideoQueue("video_freeze_queue")
+	if err != nil {
+		return err
+	}
+
+	if err = freezeBin.AddElements(element, queue); err != nil {
+		return err
+	}
+
+	// stall watchdog threshold: 2x segment duration
+	stallAfter := 8 * time.Second
+	if o := b.conf.GetSegmentConfig(); o != nil {
+		stallAfter = 2 * time.Duration(o.SegmentDuration) * time.Second
+	}
+	b.freeze = newFreezeInjector(freezeSrc, stallAfter, func() bool {
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		return b.selectedPad == videoTestSrcName
+	})
+
+	freezeBin.SetEOSFunc(func() bool {
+		b.freeze.Stop(true)
+		return false
+	})
+
+	b.createTestSrcPad()
+	return nil
+}
+
 func (b *VideoBin) addEncoder() error {
 	videoQueue, err := gstreamer.BuildQueue("video_encoder_queue", b.conf.Latency.PipelineLatency, false)
 	if err != nil {

From 69addb4ca2b5a482a3cac36a71a046a88c630523 Mon Sep 17 00:00:00 2001
From: William Alvares <curruwilla@gmail.com>
Date: Mon, 13 Jul 2026 14:25:20 -0300
Subject: [PATCH 4/5] Declare HLS target duration with headroom in pass-through
 mode

#EXTINF entries already carry the measured duration of each closed
fragment, so variable-length segments were mostly handled. But
EXT-X-TARGETDURATION was declared as the nominal segment duration, and
in pass-through mode real segments regularly exceed it (keyframe
boundaries depend on the publisher GOP + PLI round trip), violating the
HLS spec requirement that no segment's rounded duration exceed the
declared target.

Pass-through playlists now declare 2x the nominal duration, and any
segment that still exceeds the declared target logs a warning. Audited
remaining SegmentDuration uses: splitmuxsink max-size-time (split
threshold, intentional), x264 vbv-buf-capacity (encoder path only), and
upload queue sizing - none assume fixed real durations.

Player note documented in README: window sizing by segment count
(hls.js liveSyncDurationCount) sees slightly variable latency.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md                     |  3 ++-
 pkg/pipeline/sink/segments.go | 29 +++++++++++++++++++++++++++--
 2 files changed, 29 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index e71a97f7..980fd370 100644
--- a/README.md
+++ b/README.md
@@ -69,7 +69,8 @@ logging:
 template_base: can be used to host custom templates (default http://localhost:<template_port>/)
 backup_storage: files will be moved here when uploads fail. location must have write access granted for all users
 enable_chrome_sandbox: if true, egress will run Chrome with sandboxing enabled. This requires a specific Docker setup, see below.
-enable_video_passthrough: if true, track composite and participant egress with HLS segment output may skip video decoding/encoding when the published track is already H.264 and no encoding options were requested. Falls back to transcoding otherwise (default false)
+enable_video_passthrough: if true, track composite and participant egress with HLS segment output may skip video decoding/encoding when the published track is already H.264 and no encoding options were requested. Falls back to transcoding otherwise (default false).
+  Note on segment durations: in pass-through mode segment boundaries depend on the publisher's keyframe interval and the keyframe request (PLI) round trip, not on a local encoder, so real segment durations vary (typically nominal + up to one GOP + RTT). #EXTINF entries always carry the measured duration and EXT-X-TARGETDURATION is declared as 2x the nominal segment duration for spec compliance. Players that size their live window by segment count (e.g. hls.js liveSyncDurationCount) will see slightly variable display latency.
 cpu_cost: # optionally override cpu cost estimation, used when accepting or denying requests
   room_composite_cpu_cost: 3.0
   web_cpu_cost: 3.0
diff --git a/pkg/pipeline/sink/segments.go b/pkg/pipeline/sink/segments.go
index 2ab1ffba..31d3db87 100644
--- a/pkg/pipeline/sink/segments.go
+++ b/pkg/pipeline/sink/segments.go
@@ -59,6 +59,7 @@ type SegmentSink struct {
 	startTime             time.Time
 	lastUpload            time.Time
 	outputType            types.OutputType
+	targetDuration        int
 	startRunningTime      uint64
 	openSegmentsStartTime map[string]uint64
 
@@ -85,8 +86,23 @@ func newSegmentSink(
 		return nil, err
 	}
 
+	// Segments are cut at keyframes, so real durations always vary around the
+	// nominal segment duration (#EXTINF entries already use the measured
+	// duration of each closed fragment). In pass-through mode the variance is
+	// larger: boundaries depend on the publisher's GOP and the
+	// GstForceKeyUnit->PLI round trip instead of a local encoder, so segments
+	// typically run one GOP + RTT over nominal. The HLS spec requires
+	// EXT-X-TARGETDURATION >= the rounded duration of every segment, so
+	// declare it with headroom. Players that size their live window by
+	// segment count (e.g. hls.js liveSyncDurationCount) will see slightly
+	// variable display latency - acceptable for this use case.
+	targetDuration := o.SegmentDuration
+	if conf.VideoPassthrough {
+		targetDuration *= 2
+	}
+
 	playlistName := path.Join(o.LocalDir, o.PlaylistFilename)
-	playlist, err := m3u8.NewEventPlaylistWriter(playlistName, o.SegmentDuration)
+	playlist, err := m3u8.NewEventPlaylistWriter(playlistName, targetDuration)
 	if err != nil {
 		return nil, err
 	}
@@ -94,7 +110,7 @@ func newSegmentSink(
 	var livePlaylist m3u8.PlaylistWriter
 	if o.LivePlaylistFilename != "" {
 		playlistName = path.Join(o.LocalDir, o.LivePlaylistFilename)
-		livePlaylist, err = m3u8.NewLivePlaylistWriter(playlistName, o.SegmentDuration, defaultLivePlaylistWindow)
+		livePlaylist, err = m3u8.NewLivePlaylistWriter(playlistName, targetDuration, defaultLivePlaylistWindow)
 		if err != nil {
 			return nil, err
 		}
@@ -125,6 +141,7 @@ func newSegmentSink(
 		playlist:              playlist,
 		livePlaylist:          livePlaylist,
 		outputType:            outputType,
+		targetDuration:        targetDuration,
 		openSegmentsStartTime: make(map[string]uint64),
 		closedSegments:        make(chan SegmentUpdate, maxPendingUploads),
 		playlistUpdates:       make(chan SegmentUpdate, maxPendingUploads),
@@ -209,6 +226,14 @@ func (s *SegmentSink) handlePlaylistUpdates(update SegmentUpdate) error {
 	duration := float64(time.Duration(update.endTime-t)) / float64(time.Second)
 	segmentStartTime := s.startTime.Add(time.Duration(t - s.startRunningTime))
 
+	// a violation breaks spec-compliant players' buffering assumptions
+	if int(duration+0.5) > s.targetDuration {
+		logger.Warnw("segment duration exceeds declared EXT-X-TARGETDURATION", nil,
+			"filename", update.filename,
+			"duration", duration,
+			"targetDuration", s.targetDuration)
+	}
+
 	// do not update playlist until upload is complete
 	<-update.uploadComplete
 

From 51aadef2dde7f83f5ffdb605f3c649e952daf95e Mon Sep 17 00:00:00 2001
From: William Alvares <curruwilla@gmail.com>
Date: Mon, 13 Jul 2026 14:30:14 -0300
Subject: [PATCH 5/5] Add pass-through integration test scenarios and CPU
 measurement

Failure-focused staging scenarios for the pass-through feature: without
a decoder there is no loss concealment, so a corrupted bitstream reaches
h264parse/mpegtsmux directly and must degrade gracefully.

- publish.sh: synthetic OBS-like publisher (pre-encoded H.264, fixed 2s
  GOP + Opus) over WHIP bypass, with pause/resume/stop for mute and
  disconnect scenarios
- egress.sh: participant egress with HLS output; --transcode adds a
  preset to force the decode+encode fallback on the same deployment
- netem.sh: port-scoped packet loss on loopback (publisher<->ingress and
  SFU<->egress legs)
- measure-cpu.sh: container + per-handler CPU sampling to CSV for the
  30-minute pass-through vs transcode comparison
- check-avsync.sh: per-segment audio/video first-PTS offset (transcoded
  audio vs pass-through video drift detection)

README maps each script to the five scenarios and their pass criteria.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 test/passthrough/README.md       | 120 +++++++++++++++++++++++++++++++
 test/passthrough/check-avsync.sh |  51 +++++++++++++
 test/passthrough/egress.sh       |  64 +++++++++++++++++
 test/passthrough/measure-cpu.sh  |  61 ++++++++++++++++
 test/passthrough/netem.sh        |  71 ++++++++++++++++++
 test/passthrough/publish.sh      |  94 ++++++++++++++++++++++++
 6 files changed, 461 insertions(+)
 create mode 100644 test/passthrough/README.md
 create mode 100755 test/passthrough/check-avsync.sh
 create mode 100755 test/passthrough/egress.sh
 create mode 100755 test/passthrough/measure-cpu.sh
 create mode 100755 test/passthrough/netem.sh
 create mode 100755 test/passthrough/publish.sh

diff --git a/test/passthrough/README.md b/test/passthrough/README.md
new file mode 100644
index 00000000..38fb1e2b
--- /dev/null
+++ b/test/passthrough/README.md
@@ -0,0 +1,120 @@
+# Pass-through integration test scenarios
+
+Test scenarios for the H.264 video pass-through feature (branch
+`feature/egress-h264-passthrough`). These are deliberately failure-focused:
+without a decoder in the path there is no packet loss concealment, so a
+corrupted bitstream reaches h264parse/mpegtsmux directly (the issue #723 error
+class). Every scenario must degrade gracefully - a stalled playlist or a
+crashed pipeline is a failure.
+
+## Environment assumptions
+
+- A local LiveKit stack running with host networking: LiveKit server
+  (`ws://localhost:7880`, RTC UDP 50000-60000 by default), Ingress (WHIP on
+  `http://localhost:8080/w`, RTC UDP 7885), Egress built from this branch with
+  `enable_video_passthrough: true`. Adjust ports to your deployment.
+- `lk` CLI on the host.
+- `tc` (iproute2) on the host for netem, root required.
+- Scripts require `LIVEKIT_API_KEY` and `LIVEKIT_API_SECRET` in the
+  environment; `LIVEKIT_URL` defaults to `ws://localhost:7880`. Set
+  `EGRESS_CONTAINER` to the egress container name for `measure-cpu.sh`.
+
+## Scripts
+
+| Script | Purpose |
+|---|---|
+| `publish.sh` | Synthetic OBS-like publisher: pre-encoded H.264 (2s fixed GOP, zerolatency) + Opus audio over WHIP bypass. Supports `pause`/`resume`/`stop` for mute/disconnect scenarios. |
+| `egress.sh` | Starts a Participant egress with HLS segment output for the test room. `--transcode` adds an encoding preset, which forces the decode+encode fallback - same deployment, both modes. |
+| `netem.sh` | Applies/removes packet loss on a UDP port range on `lo` (host networking → both legs cross loopback). |
+| `measure-cpu.sh` | Samples the egress container CPU to CSV and prints the average. Run once in pass-through and once in `--transcode` mode for the 5.6 comparison. |
+| `check-avsync.sh` | ffprobe audio/video first-PTS offset per segment, for the A/V sync scenario. |
+
+## Scenarios
+
+### 5.1 Packet loss publisher↔ingress
+
+```bash
+./publish.sh start
+./egress.sh start
+sudo ./netem.sh apply 7885 5%       # ingress RTC port
+# observe 2-3 minutes, then:
+sudo ./netem.sh clear
+```
+
+Pass criteria: egress handler log shows PLIs being sent and possibly dropped
+buffers, but no pipeline error/crash; playlist keeps advancing (frozen or
+degraded frames acceptable); egress survives until `lk egress stop`.
+
+### 5.2 Packet loss SFU↔egress
+
+Same as 5.1 but against the SFU RTC range:
+
+```bash
+sudo ./netem.sh apply 50000-60000 5%
+```
+
+Note: the SFU range is shared with any other room traffic on the stack - run
+during a quiet window. Watch for `keyframe probe` PLI logs and jitter buffer
+drops; same pass criteria.
+
+### 5.3 Publisher mute mid-session (validates the freeze fallback)
+
+```bash
+./publish.sh start
+./egress.sh start
+sleep 60
+./publish.sh pause          # SIGSTOP - RTP stops, appwriter inactivity fires
+sleep 30
+./publish.sh resume
+```
+
+Pass criteria: within ~2s of pause the handler logs `video freeze started`;
+playlist keeps gaining segments (frozen frame) at the nominal cadence; on
+resume, `video freeze ended` + `sending PLI to publisher`, live video returns
+within one GOP + RTT. Player must not stall at any point.
+
+### 5.4 Publisher disconnect/reconnect
+
+```bash
+./publish.sh stop           # hard disconnect
+sleep 20
+./publish.sh start          # new track, same identity
+```
+
+Pass criteria: on disconnect the freeze fallback holds the playlist; on
+reconnect the new track is picked up (Participant egress resubscribes), freeze
+ends, video resumes. The pipeline must not hang permanently even if the
+publisher never returns (session limits still apply).
+
+### 5.5 A/V sync: pass-through video + transcoded audio
+
+Audio is always transcoded (Opus→AAC) while video passes through, so their
+pipeline latencies differ. After running 5.1-5.4, download a window of
+segments (start / after loss / after mute / after reconnect) and run:
+
+```bash
+./check-avsync.sh <segment-dir>
+```
+
+Pass criteria: audio/video first-PTS offset per segment stays stable (no
+cumulative drift) and within ±100ms of the session's initial offset,
+including after loss/mute events.
+
+### 5.6 CPU comparison (the point of all this)
+
+Two 30-minute runs with the same publisher settings:
+
+```bash
+./publish.sh start
+./egress.sh start                        # pass-through
+./measure-cpu.sh passthrough.csv 1800
+./egress.sh stop
+
+./egress.sh start --transcode           # preset forces decode+encode fallback
+./measure-cpu.sh transcode.csv 1800
+./egress.sh stop
+```
+
+Report the two averages. No redeploy needed between runs - the `--transcode`
+egress carries an encoding preset, which the config gate treats as a
+transcoding request.
diff --git a/test/passthrough/check-avsync.sh b/test/passthrough/check-avsync.sh
new file mode 100755
index 00000000..29adec18
--- /dev/null
+++ b/test/passthrough/check-avsync.sh
@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+# Check audio/video PTS alignment across HLS segments.
+# Pass-through video + transcoded audio (Opus->AAC) take different pipeline
+# paths; this prints the audio-video first-PTS offset per segment so drift or
+# jumps after loss/mute events are visible.
+#
+# Usage: ./check-avsync.sh <dir-with-ts-segments>
+# Requires ffprobe (falls back to a dockerized ffprobe if not installed).
+set -euo pipefail
+
+DIR="${1:?directory with .ts segments required}"
+
+ffprobe_cmd() {
+    if command -v ffprobe >/dev/null; then
+        ffprobe "$@"
+    else
+        docker run --rm -v "$(realpath "$DIR")":/data -w /data --entrypoint ffprobe \
+            linuxserver/ffmpeg "$@"
+    fi
+}
+
+printf '%-40s %12s %12s %12s\n' "segment" "video_pts" "audio_pts" "offset_ms"
+
+first_offset=""
+for f in "$DIR"/*.ts; do
+    name=$(basename "$f")
+    probe_target="$name"
+    command -v ffprobe >/dev/null && probe_target="$f"
+
+    v=$(ffprobe_cmd -v error -select_streams v:0 -show_entries packet=pts_time \
+        -of csv=p=0 -read_intervals '%+#1' "$probe_target" 2>/dev/null | head -1)
+    a=$(ffprobe_cmd -v error -select_streams a:0 -show_entries packet=pts_time \
+        -of csv=p=0 -read_intervals '%+#1' "$probe_target" 2>/dev/null | head -1)
+
+    if [[ -z "$v" || -z "$a" ]]; then
+        printf '%-40s %12s %12s %12s\n' "$name" "${v:-none}" "${a:-none}" "n/a"
+        continue
+    fi
+
+    offset=$(awk -v v="$v" -v a="$a" 'BEGIN { printf "%.1f", (a - v) * 1000 }')
+    drift=""
+    if [[ -z "$first_offset" ]]; then
+        first_offset="$offset"
+    else
+        drift=$(awk -v o="$offset" -v f="$first_offset" 'BEGIN {
+            d = o - f
+            if (d > 100 || d < -100) printf "  <-- drift %.1fms", d
+        }')
+    fi
+    printf '%-40s %12s %12s %12s%s\n' "$name" "$v" "$a" "$offset" "$drift"
+done
diff --git a/test/passthrough/egress.sh b/test/passthrough/egress.sh
new file mode 100755
index 00000000..4395ff65
--- /dev/null
+++ b/test/passthrough/egress.sh
@@ -0,0 +1,64 @@
+#!/usr/bin/env bash
+# Start/stop a Participant egress with HLS segment output for the test room.
+#
+# Usage:
+#   ./egress.sh start [--transcode] [room]   start egress (pass-through by default)
+#   ./egress.sh stop                          stop the last started egress
+#
+# --transcode adds an encoding preset to the request. The config gate treats
+# any preset as a transcoding request, so this forces the decode+encode
+# fallback on the same deployment - used for the CPU comparison (5.6).
+set -euo pipefail
+
+STATE_DIR="${TMPDIR:-/tmp}/passthrough-test"
+mkdir -p "$STATE_DIR"
+
+export LIVEKIT_URL="${LIVEKIT_URL:-ws://localhost:7880}"
+: "${LIVEKIT_API_KEY:?set LIVEKIT_API_KEY}"
+: "${LIVEKIT_API_SECRET:?set LIVEKIT_API_SECRET}"
+
+case "${1:-}" in
+start)
+    shift
+    TRANSCODE=false
+    ROOM="passthrough-test"
+    for arg in "$@"; do
+        case "$arg" in
+        --transcode) TRANSCODE=true ;;
+        *) ROOM="$arg" ;;
+        esac
+    done
+
+    PREFIX="passthrough-test/$(date +%Y%m%d-%H%M%S)"
+    REQ="$STATE_DIR/egress-request.json"
+    {
+        echo '{'
+        echo "  \"room_name\": \"$ROOM\","
+        echo '  "identity": "publisher",'
+        if $TRANSCODE; then
+            echo '  "preset": "H264_720P_30",'
+        fi
+        echo '  "segment_outputs": [{'
+        echo "    \"filename_prefix\": \"$PREFIX/seg\","
+        echo "    \"playlist_name\": \"$PREFIX/playlist.m3u8\","
+        echo "    \"live_playlist_name\": \"$PREFIX/live.m3u8\","
+        echo '    "segment_duration": 4'
+        echo '  }]'
+        echo '}'
+    } > "$REQ"
+
+    lk egress start --type participant "$REQ" | tee "$STATE_DIR/egress-start.out"
+    grep -oE 'EG_[A-Za-z0-9]+' "$STATE_DIR/egress-start.out" | head -1 > "$STATE_DIR/egress_id"
+    echo "egress id: $(cat "$STATE_DIR/egress_id"), output prefix: $PREFIX"
+    echo "mode: $($TRANSCODE && echo 'transcode (forced fallback)' || echo 'pass-through')"
+    ;;
+
+stop)
+    lk egress stop "$(cat "$STATE_DIR/egress_id")"
+    ;;
+
+*)
+    echo "usage: $0 start [--transcode] [room] | stop" >&2
+    exit 1
+    ;;
+esac
diff --git a/test/passthrough/measure-cpu.sh b/test/passthrough/measure-cpu.sh
new file mode 100755
index 00000000..a1d022c9
--- /dev/null
+++ b/test/passthrough/measure-cpu.sh
@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+# Sample egress CPU usage to CSV and print the average.
+# Measures both the container total (docker stats) and the per-handler process
+# (/proc inside the container), since one service process can host multiple
+# handler subprocesses.
+#
+# Usage: ./measure-cpu.sh <output.csv> [duration_seconds] [container]
+# Container defaults to $EGRESS_CONTAINER or "egress".
+set -euo pipefail
+
+OUT="${1:?output csv required}"
+DURATION="${2:-1800}"
+CONTAINER="${3:-${EGRESS_CONTAINER:-egress}}"
+INTERVAL=5
+
+echo "timestamp,container_cpu_pct,handler_cpu_pct,handler_pids" > "$OUT"
+
+read_handler_jiffies() {
+    # sum utime+stime for every run-handler process (and children via their own entries)
+    docker exec "$CONTAINER" sh -c '
+        total=0; pids=""
+        for pid in $(ls /proc | grep -E "^[0-9]+$"); do
+            if grep -qa "run-handler" /proc/$pid/cmdline 2>/dev/null; then
+                stat=$(cat /proc/$pid/stat 2>/dev/null) || continue
+                set -- $stat
+                total=$((total + $14 + $15))
+                pids="$pids $pid"
+            fi
+        done
+        cpu_total=$(awk "/^cpu /{sum=0; for(i=2;i<=NF;i++) sum+=\$i; print sum}" /proc/stat)
+        echo "$total $cpu_total $pids"
+    '
+}
+
+NCPU=$(docker exec "$CONTAINER" nproc)
+END=$(( $(date +%s) + DURATION ))
+prev=$(read_handler_jiffies)
+
+echo "sampling $CONTAINER every ${INTERVAL}s for ${DURATION}s -> $OUT"
+while (( $(date +%s) < END )); do
+    sleep "$INTERVAL"
+    cur=$(read_handler_jiffies)
+
+    prev_h=$(echo "$prev" | awk '{print $1}')
+    prev_t=$(echo "$prev" | awk '{print $2}')
+    cur_h=$(echo "$cur" | awk '{print $1}')
+    cur_t=$(echo "$cur" | awk '{print $2}')
+    pids=$(echo "$cur" | cut -d' ' -f3- | tr -s ' ' ';')
+
+    handler_pct=$(awk -v h=$((cur_h - prev_h)) -v t=$((cur_t - prev_t)) -v n="$NCPU" \
+        'BEGIN { if (t > 0) printf "%.2f", 100 * n * h / t; else print "0" }')
+    container_pct=$(docker stats --no-stream --format '{{.CPUPerc}}' "$CONTAINER" | tr -d '%')
+
+    echo "$(date +%s),$container_pct,$handler_pct,$pids" >> "$OUT"
+    prev="$cur"
+done
+
+echo "--- summary ($OUT) ---"
+awk -F, 'NR>1 { c+=$2; h+=$3; n++ } END {
+    if (n > 0) printf "samples: %d\navg container cpu: %.2f%%\navg handler cpu: %.2f%%\n", n, c/n, h/n
+}' "$OUT"
diff --git a/test/passthrough/netem.sh b/test/passthrough/netem.sh
new file mode 100755
index 00000000..52485bec
--- /dev/null
+++ b/test/passthrough/netem.sh
@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+# Apply/remove packet loss on a UDP port (or range) on the loopback interface.
+# With host networking every leg (publisher->ingress, SFU->egress) crosses lo,
+# so loss is scoped by port to avoid disturbing redis/API traffic.
+#
+# Usage:
+#   sudo ./netem.sh apply 7885 5%             # single port (ingress RTC)
+#   sudo ./netem.sh apply 50000-60000 5%      # port range (SFU RTC)
+#   sudo ./netem.sh clear
+#   ./netem.sh status
+set -euo pipefail
+
+DEV="${DEV:-lo}"
+
+case "${1:-}" in
+apply)
+    PORTS="${2:?port or port range required}"
+    LOSS="${3:-5%}"
+
+    # root: prio qdisc, band 3 gets netem loss; filters select the target ports
+    tc qdisc replace dev "$DEV" root handle 1: prio bands 4
+    tc qdisc replace dev "$DEV" parent 1:4 handle 40: netem loss "$LOSS"
+
+    add_filters() {
+        local port="$1"
+        # match both directions
+        tc filter add dev "$DEV" protocol ip parent 1:0 prio 1 u32 \
+            match ip protocol 17 0xff match ip dport "$port" 0xffff flowid 1:4
+        tc filter add dev "$DEV" protocol ip parent 1:0 prio 1 u32 \
+            match ip protocol 17 0xff match ip sport "$port" 0xffff flowid 1:4
+    }
+
+    if [[ "$PORTS" == *-* ]]; then
+        START="${PORTS%-*}"
+        END="${PORTS#*-}"
+        # u32 has no native range match; use a mask-aligned walk
+        port=$START
+        while (( port <= END )); do
+            # find the largest power-of-two block starting at $port within range
+            span=1
+            while (( (port % (span * 2)) == 0 && port + span * 2 - 1 <= END )); do
+                span=$((span * 2))
+            done
+            mask=$(printf '0x%04x' $(( 0xffff & ~(span - 1) )))
+            tc filter add dev "$DEV" protocol ip parent 1:0 prio 1 u32 \
+                match ip protocol 17 0xff match ip dport "$port" "$mask" flowid 1:4
+            tc filter add dev "$DEV" protocol ip parent 1:0 prio 1 u32 \
+                match ip protocol 17 0xff match ip sport "$port" "$mask" flowid 1:4
+            port=$((port + span))
+        done
+    else
+        add_filters "$PORTS"
+    fi
+    echo "netem loss $LOSS applied on $DEV udp $PORTS at $(date +%s.%N)"
+    ;;
+
+clear)
+    tc qdisc del dev "$DEV" root 2>/dev/null || true
+    echo "netem cleared on $DEV at $(date +%s.%N)"
+    ;;
+
+status)
+    tc -s qdisc show dev "$DEV"
+    tc filter show dev "$DEV" 2>/dev/null | head -20
+    ;;
+
+*)
+    echo "usage: $0 apply <port|start-end> [loss%] | clear | status" >&2
+    exit 1
+    ;;
+esac
diff --git a/test/passthrough/publish.sh b/test/passthrough/publish.sh
new file mode 100755
index 00000000..eb41f4d4
--- /dev/null
+++ b/test/passthrough/publish.sh
@@ -0,0 +1,94 @@
+#!/usr/bin/env bash
+# Synthetic OBS-like publisher for pass-through testing: pre-encoded H.264
+# (fixed 2s GOP, zerolatency) + Opus audio, delivered over WHIP bypass.
+#
+# Usage:
+#   ./publish.sh start [room]   create WHIP ingress (bypass) + start publisher
+#   ./publish.sh pause          SIGSTOP the publisher (mute-like: RTP stops)
+#   ./publish.sh resume         SIGCONT the publisher
+#   ./publish.sh stop           kill publisher container (disconnect)
+set -euo pipefail
+
+ROOM="${2:-passthrough-test}"
+CONTAINER="passthrough-publisher"
+STATE_DIR="${TMPDIR:-/tmp}/passthrough-test"
+GST_IMAGE="${GST_IMAGE:-livekit/gstreamer:1.24.12-dev}"
+
+export LIVEKIT_URL="${LIVEKIT_URL:-ws://localhost:7880}"
+: "${LIVEKIT_API_KEY:?set LIVEKIT_API_KEY}"
+: "${LIVEKIT_API_SECRET:?set LIVEKIT_API_SECRET}"
+
+mkdir -p "$STATE_DIR"
+
+case "${1:-}" in
+start)
+    # WHIP_URL env overrides ingress creation (e.g. reuse an existing ingress)
+    if [[ -n "${WHIP_URL:-}" ]]; then
+        echo "$WHIP_URL" > "$STATE_DIR/whip_url"
+    fi
+    # reuse ingress across reconnects so the stream key stays stable
+    if [[ ! -f "$STATE_DIR/whip_url" ]]; then
+        cat > "$STATE_DIR/ingress-request.json" <<EOF
+{
+  "input_type": "WHIP_INPUT",
+  "name": "passthrough-test",
+  "room_name": "$ROOM",
+  "participant_identity": "publisher",
+  "enable_transcoding": false
+}
+EOF
+        lk ingress create "$STATE_DIR/ingress-request.json" > "$STATE_DIR/ingress.out"
+        cat "$STATE_DIR/ingress.out"
+        URL=$(grep -oE 'https?://[^" ]+/w[^" ]*' "$STATE_DIR/ingress.out" | head -1)
+        KEY=$(grep -oE '"?stream_?[kK]ey"?[": ]+[A-Za-z0-9_-]+' "$STATE_DIR/ingress.out" | grep -oE '[A-Za-z0-9_-]+$' | head -1)
+        echo "${URL%/}/${KEY}" > "$STATE_DIR/whip_url"
+    fi
+    WHIP_URL=$(cat "$STATE_DIR/whip_url")
+    echo "publishing to $WHIP_URL"
+
+    docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+    # 30fps, key-int-max=60 => fixed 2s GOP like the OBS setup
+    docker run -d --name "$CONTAINER" --network host "$GST_IMAGE" \
+        gst-launch-1.0 -e \
+        videotestsrc is-live=true pattern=smpte \
+            ! video/x-raw,width=1280,height=720,framerate=30/1 \
+            ! timeoverlay font-desc="Sans 24" \
+            ! x264enc tune=zerolatency speed-preset=veryfast key-int-max=60 bitrate=2500 option-string="scenecut=0" \
+            ! h264parse \
+            ! whip.video_0 \
+        audiotestsrc is-live=true wave=sine freq=440 \
+            ! audio/x-raw,rate=48000,channels=2 \
+            ! opusenc \
+            ! whip.audio_0 \
+        whipclientsink name=whip signaller::whip-endpoint="$WHIP_URL"
+    echo "publisher started (container $CONTAINER)"
+    ;;
+
+pause)
+    docker kill --signal SIGSTOP "$CONTAINER"
+    echo "publisher paused (RTP stopped) at $(date +%s.%N)"
+    ;;
+
+resume)
+    docker kill --signal SIGCONT "$CONTAINER"
+    echo "publisher resumed at $(date +%s.%N)"
+    ;;
+
+stop)
+    docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+    echo "publisher disconnected at $(date +%s.%N)"
+    ;;
+
+clean)
+    docker rm -f "$CONTAINER" >/dev/null 2>&1 || true
+    if [[ -f "$STATE_DIR/ingress.json" ]]; then
+        lk ingress delete "$(jq -r '.ingress_id' "$STATE_DIR/ingress.json")" || true
+    fi
+    rm -rf "$STATE_DIR"
+    ;;
+
+*)
+    echo "usage: $0 start [room] | pause | resume | stop | clean" >&2
+    exit 1
+    ;;
+esac