Gate late sinks on live keyframes

Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>

Author: minpeter

internal/airplay/capture_broadcast.go

@@ -96,6 +96,7 @@ type BroadcastSink struct {
 	maxFrameQueueDuration time.Duration
 	backpressure          bool
 	blockedProducers      int // number waiting for queue handoff; guarded by mu
+	awaitingRandomAccess  bool
 
 	inputClosed   bool // the source ended; drain queue, then return EOF
 	closed        bool // explicitly removed; discard queue and return EOF
@@ -166,6 +167,7 @@ func (bc *BroadcastCapture) AddSink() *BroadcastSink {
 	s.primer = bc.primer
 	if len(s.primer.AnnexB) > 0 {
 		s.primer.PTS = bc.now()
+		s.awaitingRandomAccess = true
 	}
 	bc.sinks = append(bc.sinks, s)
 	bc.mu.Unlock()
@@ -195,6 +197,7 @@ func (bc *BroadcastCapture) AddBackpressuredSink() (*BroadcastSink, error) {
 	s.primer = bc.primer
 	if len(s.primer.AnnexB) > 0 {
 		s.primer.PTS = bc.now()
+		s.awaitingRandomAccess = true
 	}
 	bc.sinks = append(bc.sinks, s)
 	bc.mu.Unlock()
@@ -343,6 +346,25 @@ func isDecoderPrimer(annexB []byte) bool {
 	return h264SPS && h264PPS && h264IDR || hevcVPS && hevcSPS && hevcPPS && hevcIRAP
 }
 
+func isRandomAccessUnit(annexB []byte) bool {
+	for _, nal := range splitAnnexBAccessUnit(annexB) {
+		raw := stripStartCode(nal)
+		if len(raw) == 0 {
+			continue
+		}
+		if raw[0]&0x1f == 5 {
+			return true
+		}
+		if len(raw) >= 2 {
+			nalType := hevcNALType(raw)
+			if nalType >= 16 && nalType <= 23 {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // finish stops accepting sinks, lets existing sinks drain, and only then
 // publishes BroadcastCapture completion.
 func (bc *BroadcastCapture) finish(err error) {
@@ -456,6 +478,12 @@ func (s *BroadcastSink) enqueueFrame(frame VideoAccessUnit) error {
 		if s.closed || s.inputClosed {
 			return io.ErrClosedPipe
 		}
+		if s.awaitingRandomAccess {
+			if !isRandomAccessUnit(frame.AnnexB) {
+				return nil
+			}
+			s.awaitingRandomAccess = false
+		}
 		if len(s.frameQueue) == 0 && len(frame.AnnexB) > s.maxQueuedBytes {
 			return errBroadcastSinkBacklog
 		}

internal/airplay/capture_broadcast_test.go

@@ -181,14 +181,20 @@ func TestBroadcastCaptureReplaysDecoderPrimerToLateSink(t *testing.T) {
 		PTS:    boundary.PTS.Add(time.Second / 30),
 	}
 	frames <- live
+	<-reads
+	randomAccess := VideoAccessUnit{
+		AnnexB: []byte{0, 0, 0, 1, 0x65, 0x80},
+		PTS:    live.PTS.Add(time.Second / 30),
+	}
+	frames <- randomAccess
 	close(frames)
 
 	next, err := sink.ReadVideoAccessUnit()
 	if err != nil {
 		t.Fatalf("read live frame after decoder primer: %v", err)
 	}
-	if !bytes.Equal(next.AnnexB, live.AnnexB) || !next.PTS.Equal(live.PTS) {
-		t.Fatalf("live frame after primer = {%x %v}, want {%x %v}", next.AnnexB, next.PTS, live.AnnexB, live.PTS)
+	if !bytes.Equal(next.AnnexB, randomAccess.AnnexB) || !next.PTS.Equal(randomAccess.PTS) {
+		t.Fatalf("live frame after primer = {%x %v}, want random access {%x %v}", next.AnnexB, next.PTS, randomAccess.AnnexB, randomAccess.PTS)
 	}
 	if err := <-runDone; !errors.Is(err, io.EOF) {
 		t.Fatalf("broadcast run = %v, want EOF", err)