openvibers/openvibe.link Commits

Video: WHIP publisher, test pattern, command and JPEG sources

OP OpenVibers committed 29 Sept 2026 · 70f62c5eb0 · 9 files, +1613 −23

pion/webrtc publishes H.264 to the whip_url with the publish key as bearer,
answers PLI/FIR with a keyframe, reconnects with backoff and DELETEs the
session on shutdown. A small PCM/skip H.264 encoder draws the dry-run test
pattern without any codec library (verified by decoding with libavcodec);
it is also the no-ffmpeg fallback for plugin JPEG frames. Child processes
(rpicam-vid, ffmpeg) are split into access units and restarted when they
exit. Plugin.Begin queues a command in arrival order and waits separately.

Changes

ddiff --git a/go.mod b/go.mod
iindex c2d0011..75168f3 100644
---- a/go.mod
++++ b/go.mod
@@@ -2,4 +2,31 @@ module github.com/OpenVibers/OpenVibe.Node
  
  go 1.24.7
  
--require github.com/gorilla/websocket v1.5.3
++require (
++	github.com/gorilla/websocket v1.5.3
++	github.com/pion/interceptor v0.1.49
++	github.com/pion/rtcp v1.2.18
++	github.com/pion/webrtc/v4 v4.2.22
++)
++
++require (
++	github.com/google/uuid v1.6.0 // indirect
++	github.com/pion/datachannel v1.6.3 // indirect
++	github.com/pion/dtls/v3 v3.1.9 // indirect
++	github.com/pion/ice/v4 v4.4.4 // indirect
++	github.com/pion/logging v0.2.4 // indirect
++	github.com/pion/mdns/v2 v2.2.1 // indirect
++	github.com/pion/randutil v0.1.0 // indirect
++	github.com/pion/rtp v1.10.5 // indirect
++	github.com/pion/sctp v1.11.3 // indirect
++	github.com/pion/sdp/v3 v3.0.20 // indirect
++	github.com/pion/srtp/v3 v3.1.0 // indirect
++	github.com/pion/stun/v4 v4.0.1 // indirect
++	github.com/pion/transport/v5 v5.1.1 // indirect
++	github.com/pion/turn/v5 v5.1.2 // indirect
++	github.com/wlynxg/anet v0.0.5 // indirect
++	golang.org/x/crypto v0.48.0 // indirect
++	golang.org/x/net v0.50.0 // indirect
++	golang.org/x/sys v0.41.0 // indirect
++	golang.org/x/time v0.14.0 // indirect
++)
ddiff --git a/go.sum b/go.sum
iindex 25a9fc4..f8ff03c 100644
---- a/go.sum
++++ b/go.sum
@@@ -1,2 +1,50 @@
++github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
++github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
  github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
  github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
++github.com/pion/datachannel v1.6.3 h1:HZAYLpunVI0xi057ZiANl558d0uSpomcWw4UEPz3qXs=
++github.com/pion/datachannel v1.6.3/go.mod h1:oRmkXVVXxkUbqa8MatgdNHW29kvtQY3HUdjJQsfv2GE=
++github.com/pion/dtls/v3 v3.1.9 h1:rpeycmLIkc4krpk1IxP7+39o11QdCXbmV9+FGi9yZJ8=
++github.com/pion/dtls/v3 v3.1.9/go.mod h1:iKFQNYrjsN2TiA2YKKMqB9MOZaFpjFULBI/A4sW0eyc=
++github.com/pion/ice/v4 v4.4.4 h1:nPhqLkf8NoE9O8NBKd3Xo2NNc2PlBMdM07tWZ+IH/i8=
++github.com/pion/ice/v4 v4.4.4/go.mod h1:vkrbLwF98P/DJuCKoAmOQY4FJ7dJPI5VcBFOO3kyYx0=
++github.com/pion/interceptor v0.1.49 h1:iyBsNHRoLNNXZiJA/DdGvJgdyfS8YWDCzxIAlVjt5nY=
++github.com/pion/interceptor v0.1.49/go.mod h1:MZ6PJkja/TCo350HAnBrs/rUIyad9mWjpcvytrf3ViQ=
++github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8=
++github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so=
++github.com/pion/mdns/v2 v2.2.1 h1:lQPWv6Mo5jG1sKwqkYgfeXGX7/seban4x7/j0n485/M=
++github.com/pion/mdns/v2 v2.2.1/go.mod h1:ZX5f0AAH1D6TOjdjvcBORZZHaZuG0t9+br78lHEwiJ4=
++github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
++github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
++github.com/pion/rtcp v1.2.18 h1:ZPgMZMCtBoReTixtFAyzcOkuJiebyB94qxtBAyvSooM=
++github.com/pion/rtcp v1.2.18/go.mod h1:vX7Es6skWYm4oVvOSC5+4WHLYUTZHcbGNmk5wjzMGH4=
++github.com/pion/rtp v1.10.5 h1:ip0HhO/wYZqQ4bKS+R99KnZh/GRCmIT0jDXikub7vlE=
++github.com/pion/rtp v1.10.5/go.mod h1:Au8fc6cEByy8RLTwKTQTEeQqDB/SJDxwL4mZuxYA5Pk=
++github.com/pion/sctp v1.11.3 h1:cFo/voNuKzdxEno6z96PWq/96xkAJeMB/eqB7GNfmlQ=
++github.com/pion/sctp v1.11.3/go.mod h1:7v/MXGBROc8fu9g2TVwCXRasPISCMd3j8wSH+o45uCI=
++github.com/pion/sdp/v3 v3.0.20 h1:TS6DViqcmp+49f0+mjw9anbr9xY3vJtsZewxAvlMCRQ=
++github.com/pion/sdp/v3 v3.0.20/go.mod h1:slIMXDK5OKj0nhISwjfeN18AzTBCt2LYZq9uPw0cU5Q=
++github.com/pion/srtp/v3 v3.1.0 h1:keugrBOxtvAv5Urf4lBMfso8OCAX3mm7Brdl26JSvik=
++github.com/pion/srtp/v3 v3.1.0/go.mod h1:RlTlj08MtRReQiVm/PMmMFUDNjZkoKuKJMHkl9+M7a8=
++github.com/pion/stun/v4 v4.0.1 h1:S2ggK4hsUJJKoZfpL90uIqIRTj6xJ0NOIEkXgcpQif0=
++github.com/pion/stun/v4 v4.0.1/go.mod h1:byktbPA8U7HjJ2H8w3moP7nzq0Q+kYNulvMLPaubQL4=
++github.com/pion/transport/v5 v5.1.1 h1:Sauq8YJqRopgRfIgnpfKpdpKG14g0N4jWlRgpWsEfN8=
++github.com/pion/transport/v5 v5.1.1/go.mod h1:Qxw6fCEjFWQkRDZOhS4Vf+neJBcihauvA3uyEa1J1F0=
++github.com/pion/turn/v5 v5.1.2 h1:acEOO+D5txJ52vKfcilmSMItfP23Q9K6pJ1LM5P2VV0=
++github.com/pion/turn/v5 v5.1.2/go.mod h1:57Oadc8fHnac15VFyCd6nEaBTVZBxQe2POUMD4lsHRA=
++github.com/pion/webrtc/v4 v4.2.22 h1:eoRTHzLKW5tI9o5iahw3s4KL4IHPESsfIF06zWi3JAo=
++github.com/pion/webrtc/v4 v4.2.22/go.mod h1:JBu1ba8lUEX4jbYFKofglMQJG7f0rJk7UepKjI3GaLE=
++github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
++github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
++github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
++github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
++go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
++go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
++golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts=
++golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos=
++golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60=
++golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM=
++golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
++golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
++golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
++golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
ddiff --git a/internal/plugins/plugin.go b/internal/plugins/plugin.go
iindex 54ecf42..0413e40 100644
---- a/internal/plugins/plugin.go
++++ b/internal/plugins/plugin.go
@@@ -199,48 +199,60 @@ func (p *Plugin) Resume() {
  
  // Command sends a command and waits for the plugin's ack or nack.
  func (p *Plugin) Command(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) Reply {
++	return p.Begin(ctx, id, kind, value, deadlineMS)()
++}
++
++// Begin queues a command for the plugin at once (so commands reach it in arrival order) and returns a function that
++// waits for its ack or nack. A plugin that does not answer in max(deadline, MinReplyTimeout) is told to stop and the
++// command fails with plugin_timeout.
++func (p *Plugin) Begin(ctx context.Context, id, kind string, value json.RawMessage, deadlineMS int) func() Reply {
++	now := func(r Reply) func() Reply { return func() Reply { return r } }
  	p.mu.Lock()
  	if p.state != StateReady && kind != protocol.KindHalt {
  		st, fault := p.state, p.fault
  		p.mu.Unlock()
  		if st == StateFaulted && fault != nil {
--			return Reply{FaultCode: fault.Code, Message: fault.Message}
++			return now(Reply{FaultCode: fault.Code, Message: fault.Message})
  		}
  		if st == StateDown || st == StateStopped {
--			return Reply{FaultCode: protocol.FaultPluginDown}
++			return now(Reply{FaultCode: protocol.FaultPluginDown})
  		}
--		return Reply{FaultCode: protocol.FaultNotReady}
++		return now(Reply{FaultCode: protocol.FaultNotReady})
  	}
  	ch := make(chan Reply, 1)
  	p.pending[id] = ch
  	p.mu.Unlock()
--	defer func() {
++	done := func() {
  		p.mu.Lock()
  		delete(p.pending, id)
  		p.mu.Unlock()
--	}()
++	}
  	msg := map[string]any{"op": "command", "id": id, "kind": kind, "deadline_ms": deadlineMS}
  	if len(value) > 0 {
  		msg["value"] = value
  	}
  	if err := p.send(msg); err != nil {
--		return Reply{FaultCode: protocol.FaultPluginDown}
--	}
--	wait := time.Duration(deadlineMS) * time.Millisecond
--	if wait < MinReplyTimeout {
--		wait = MinReplyTimeout
--	}
--	t := time.NewTimer(wait)
--	defer t.Stop()
--	select {
--	case r := <-ch:
--		return r
--	case <-t.C:
--		p.Stop()
--		return Reply{FaultCode: protocol.FaultPluginTimeout}
--	case <-ctx.Done():
--		p.Stop()
--		return Reply{FaultCode: protocol.FaultShuttingDown}
++		done()
++		return now(Reply{FaultCode: protocol.FaultPluginDown})
++	}
++	return func() Reply {
++		defer done()
++		wait := time.Duration(deadlineMS) * time.Millisecond
++		if wait < MinReplyTimeout {
++			wait = MinReplyTimeout
++		}
++		t := time.NewTimer(wait)
++		defer t.Stop()
++		select {
++		case r := <-ch:
++			return r
++		case <-t.C:
++			p.Stop()
++			return Reply{FaultCode: protocol.FaultPluginTimeout}
++		case <-ctx.Done():
++			p.Stop()
++			return Reply{FaultCode: protocol.FaultShuttingDown}
++		}
  	}
  }
  
ddiff --git a/internal/video/h264enc.go b/internal/video/h264enc.go
nnew file mode 100644
iindex 0000000..bcf65e0
---- /dev/null
++++ b/internal/video/h264enc.go
@@@ -0,0 +1,288 @@
++package video
++
++// A minimal H.264 encoder that needs no codec library: Constrained Baseline, CAVLC, every coded macroblock is I_PCM
++// (raw samples) and every other macroblock of a P frame is P_Skip (copied from the previous frame). It is not
++// efficient, but it is small, exact and decodable everywhere, which is what the dry-run test pattern and the
++// no-ffmpeg fallback for JPEG frames need: only the macroblocks that changed are sent after the first frame.
++//
++// The SPS advertises profile-level-id 42e01f, the value WebRTC endpoints expect.
++
++import "errors"
++
++// bitWriter writes big-endian bit strings with Exp-Golomb helpers (H.264 §7.2, §9.1).
++type bitWriter struct {
++	buf []byte
++	cur byte
++	n   uint8
++}
++
++func (w *bitWriter) bit(b uint) {
++	w.cur = w.cur<<1 | byte(b&1)
++	w.n++
++	if w.n == 8 {
++		w.buf = append(w.buf, w.cur)
++		w.cur, w.n = 0, 0
++	}
++}
++
++func (w *bitWriter) bits(v uint64, n int) {
++	for i := n - 1; i >= 0; i-- {
++		w.bit(uint(v>>uint(i)) & 1)
++	}
++}
++
++func (w *bitWriter) ue(v uint32) {
++	x := uint64(v) + 1
++	lz := 0
++	for t := x; t > 1; t >>= 1 {
++		lz++
++	}
++	w.bits(0, lz)
++	w.bits(x, lz+1)
++}
++
++func (w *bitWriter) se(v int32) {
++	if v > 0 {
++		w.ue(uint32(2*v - 1))
++	} else {
++		w.ue(uint32(-2 * v))
++	}
++}
++
++func (w *bitWriter) aligned() bool { return w.n == 0 }
++
++func (w *bitWriter) alignZero() {
++	for !w.aligned() {
++		w.bit(0)
++	}
++}
++
++// byteAligned appends whole bytes; the writer must be aligned.
++func (w *bitWriter) raw(b []byte) { w.buf = append(w.buf, b...) }
++
++// trailing writes rbsp_trailing_bits: a one, then zeros to the byte boundary.
++func (w *bitWriter) trailing() []byte {
++	w.bit(1)
++	w.alignZero()
++	return w.buf
++}
++
++// nal wraps an RBSP in an Annex B NAL unit with emulation prevention.
++func nal(refIdc, typ byte, rbsp []byte) []byte {
++	out := make([]byte, 0, len(rbsp)+len(rbsp)/64+5)
++	out = append(out, 0, 0, 0, 1, refIdc<<5|typ)
++	zeros := 0
++	for _, b := range rbsp {
++		if zeros >= 2 && b <= 3 {
++			out = append(out, 3)
++			zeros = 0
++		}
++		out = append(out, b)
++		if b == 0 {
++			zeros++
++		} else {
++			zeros = 0
++		}
++	}
++	return out
++}
++
++// Frame is a YUV 4:2:0 picture whose width and height are multiples of 16.
++type Frame struct {
++	W, H      int
++	Y, Cb, Cr []byte
++}
++
++func NewFrame(w, h int) *Frame {
++	return &Frame{W: w, H: h, Y: make([]byte, w*h), Cb: make([]byte, w*h/4), Cr: make([]byte, w*h/4)}
++}
++
++// PCMEncoder encodes Frames.
++type PCMEncoder struct {
++	W, H     int
++	mbw, mbh int
++	frameNum uint32
++	idrID    uint32
++	started  bool
++}
++
++func NewPCMEncoder(w, h int) (*PCMEncoder, error) {
++	if w <= 0 || h <= 0 || w%16 != 0 || h%16 != 0 {
++		return nil, errors.New("video: width and height must be positive multiples of 16")
++	}
++	return &PCMEncoder{W: w, H: h, mbw: w / 16, mbh: h / 16}, nil
++}
++
++// MBs is the number of macroblocks per frame.
++func (e *PCMEncoder) MBs() int { return e.mbw * e.mbh }
++
++func (e *PCMEncoder) sps() []byte {
++	w := &bitWriter{}
++	w.bits(66, 8)   // profile_idc: Baseline
++	w.bits(0xe0, 8) // constraint_set0..2 = 1 (Constrained Baseline), reserved zeros
++	w.bits(31, 8)   // level_idc 3.1
++	w.ue(0)         // seq_parameter_set_id
++	w.ue(0)         // log2_max_frame_num_minus4 → frame_num has 4 bits
++	w.ue(2)         // pic_order_cnt_type 2: output order = decode order
++	w.ue(1)         // max_num_ref_frames
++	w.bit(0)        // gaps_in_frame_num_value_allowed_flag
++	w.ue(uint32(e.mbw - 1))
++	w.ue(uint32(e.mbh - 1))
++	w.bit(1) // frame_mbs_only_flag
++	w.bit(1) // direct_8x8_inference_flag
++	w.bit(0) // frame_cropping_flag
++	w.bit(0) // vui_parameters_present_flag
++	return nal(3, 7, w.trailing())
++}
++
++func (e *PCMEncoder) pps() []byte {
++	w := &bitWriter{}
++	w.ue(0)  // pic_parameter_set_id
++	w.ue(0)  // seq_parameter_set_id
++	w.bit(0) // entropy_coding_mode_flag: CAVLC
++	w.bit(0) // bottom_field_pic_order_in_frame_present_flag
++	w.ue(0)  // num_slice_groups_minus1
++	w.ue(0)  // num_ref_idx_l0_default_active_minus1
++	w.ue(0)  // num_ref_idx_l1_default_active_minus1
++	w.bit(0) // weighted_pred_flag
++	w.bits(0, 2)
++	w.se(0)  // pic_init_qp_minus26
++	w.se(0)  // pic_init_qs_minus26
++	w.se(0)  // chroma_qp_index_offset
++	w.bit(1) // deblocking_filter_control_present_flag (so slices can turn it off)
++	w.bit(0) // constrained_intra_pred_flag
++	w.bit(0) // redundant_pic_cnt_present_flag
++	return nal(3, 8, w.trailing())
++}
++
++func (e *PCMEncoder) pcm(w *bitWriter, f *Frame, mb int) {
++	mx, my := mb%e.mbw, mb/e.mbw
++	w.alignZero() // pcm_alignment_zero_bit
++	for y := 0; y < 16; y++ {
++		off := (my*16+y)*f.W + mx*16
++		w.raw(f.Y[off : off+16])
++	}
++	cw := f.W / 2
++	for _, plane := range [][]byte{f.Cb, f.Cr} {
++		for y := 0; y < 8; y++ {
++			off := (my*8+y)*cw + mx*8
++			w.raw(plane[off : off+8])
++		}
++	}
++}
++
++// Encode returns one access unit in Annex B. idr (or the first frame) codes every macroblock and repeats SPS/PPS;
++// otherwise only the macroblocks listed in coded (ascending addresses) are sent and the rest are skipped.
++func (e *PCMEncoder) Encode(f *Frame, idr bool, coded []int) []byte {
++	if f.W != e.W || f.H != e.H {
++		panic("video: frame size does not match the encoder")
++	}
++	if !e.started {
++		idr = true
++	}
++	e.started = true
++	w := &bitWriter{}
++	w.ue(0) // first_mb_in_slice
++	if idr {
++		e.frameNum = 0
++		w.ue(7) // slice_type I (all slices of the picture)
++	} else {
++		w.ue(5) // slice_type P
++	}
++	w.ue(0)                           // pic_parameter_set_id
++	w.bits(uint64(e.frameNum&0xf), 4) // frame_num
++	if idr {
++		w.ue(e.idrID & 0xffff) // idr_pic_id
++		e.idrID++
++	} else {
++		w.bit(0) // num_ref_idx_active_override_flag
++		w.bit(0) // ref_pic_list_modification_flag_l0
++	}
++	// dec_ref_pic_marking (nal_ref_idc != 0)
++	if idr {
++		w.bit(0) // no_output_of_prior_pics_flag
++		w.bit(0) // long_term_reference_flag
++	} else {
++		w.bit(0) // adaptive_ref_pic_marking_mode_flag
++	}
++	w.se(0) // slice_qp_delta
++	w.ue(1) // disable_deblocking_filter_idc: off (PCM needs none, and skips copy exactly)
++
++	if idr {
++		for mb := 0; mb < e.MBs(); mb++ {
++			w.ue(25) // mb_type I_PCM
++			e.pcm(w, f, mb)
++		}
++	} else {
++		next := 0
++		for _, mb := range coded {
++			if mb < next || mb >= e.MBs() {
++				continue
++			}
++			w.ue(uint32(mb - next)) // mb_skip_run
++			w.ue(30)                // mb_type: intra in a P slice, 5 + I_PCM(25)
++			e.pcm(w, f, mb)
++			next = mb + 1
++		}
++		if rest := e.MBs() - next; rest > 0 {
++			w.ue(uint32(rest))
++		}
++	}
++	slice := w.trailing()
++	e.frameNum = (e.frameNum + 1) & 0xf
++	var au []byte
++	if idr {
++		au = append(au, e.sps()...)
++		au = append(au, e.pps()...)
++		return append(au, nal(3, 5, slice)...)
++	}
++	return append(au, nal(2, 1, slice)...)
++}
++
++// ChangedMBs lists the macroblocks whose samples differ between a and b by more than threshold (mean absolute luma
++// difference per sample). A nil a means everything changed.
++func ChangedMBs(a, b *Frame, threshold int) []int {
++	mbw, mbh := b.W/16, b.H/16
++	var out []int
++	for my := 0; my < mbh; my++ {
++		for mx := 0; mx < mbw; mx++ {
++			if a == nil {
++				out = append(out, my*mbw+mx)
++				continue
++			}
++			sum := 0
++			for y := 0; y < 16; y++ {
++				off := (my*16+y)*b.W + mx*16
++				for x := 0; x < 16; x++ {
++					d := int(a.Y[off+x]) - int(b.Y[off+x])
++					if d < 0 {
++						d = -d
++					}
++					sum += d
++				}
++			}
++			if sum > threshold*256 || chromaDiff(a, b, mx, my) > threshold*64 {
++				out = append(out, my*mbw+mx)
++			}
++		}
++	}
++	return out
++}
++
++func chromaDiff(a, b *Frame, mx, my int) int {
++	cw := b.W / 2
++	sum := 0
++	for _, p := range [][2][]byte{{a.Cb, b.Cb}, {a.Cr, b.Cr}} {
++		for y := 0; y < 8; y++ {
++			off := (my*8+y)*cw + mx*8
++			for x := 0; x < 8; x++ {
++				d := int(p[0][off+x]) - int(p[1][off+x])
++				if d < 0 {
++					d = -d
++				}
++				sum += d
++			}
++		}
++	}
++	return sum
++}
ddiff --git a/internal/video/h264enc_test.go b/internal/video/h264enc_test.go
nnew file mode 100644
iindex 0000000..4b04bb2
---- /dev/null
++++ b/internal/video/h264enc_test.go
@@@ -0,0 +1,189 @@
++package video
++
++import (
++	"bytes"
++	"context"
++	"image"
++	"image/jpeg"
++	"os"
++	"os/exec"
++	"path/filepath"
++	"testing"
++	"time"
++)
++
++func TestExpGolomb(t *testing.T) {
++	w := &bitWriter{}
++	for _, v := range []uint32{0, 1, 2, 3, 7} {
++		w.ue(v)
++	}
++	// 1 010 011 00100 0001000 → 1010 0110 0100 0001 000 + pad
++	got := w.trailing()
++	want := []byte{0xa6, 0x41, 0x01} // ...0001000 then stop bit 1 → 0000 0001 (with trailing)
++	if !bytes.Equal(got[:2], want[:2]) {
++		t.Fatalf("got %x", got)
++	}
++	w = &bitWriter{}
++	w.se(1)
++	w.se(-1)
++	w.se(2) // 010 011 00100, then the stop bit
++	if b := w.trailing(); b[0] != 0x4c || b[1] != 0x90 {
++		t.Fatalf("se %x", b)
++	}
++}
++
++func TestEmulationPrevention(t *testing.T) {
++	n := nal(3, 5, []byte{0, 0, 1, 0, 0, 0, 0, 0, 3})
++	want := []byte{0, 0, 0, 1, 0x65, 0, 0, 3, 1, 0, 0, 3, 0, 0, 3, 0, 3}
++	if !bytes.Equal(n, want) {
++		t.Fatalf("got % x", n)
++	}
++}
++
++func TestAccessUnitsRoundTrip(t *testing.T) {
++	enc, _ := NewPCMEncoder(64, 48)
++	var stream []byte
++	var prev *Frame
++	for i := 0; i < 5; i++ {
++		f := NewFrame(64, 48)
++		DrawPattern(f, i)
++		var coded []int
++		if prev != nil {
++			coded = ChangedMBs(prev, f, 0)
++		}
++		stream = append(stream, enc.Encode(f, false, coded)...)
++		prev = f
++	}
++	var keys []bool
++	err := AccessUnits(bytes.NewReader(stream), func(au []byte, key bool) { keys = append(keys, key) })
++	if err != nil || len(keys) != 5 || !keys[0] || keys[1] {
++		t.Fatalf("%v %v", err, keys)
++	}
++}
++
++func TestChangedMBs(t *testing.T) {
++	a, b := NewFrame(32, 32), NewFrame(32, 32)
++	b.Y[16*32+20] = 200 // macroblock (1,1) → address 3
++	if got := ChangedMBs(a, b, 0); len(got) != 1 || got[0] != 3 {
++		t.Fatalf("%v", got)
++	}
++	if got := ChangedMBs(nil, b, 0); len(got) != 4 {
++		t.Fatalf("%v", got)
++	}
++}
++
++func TestTestPatternRuns(t *testing.T) {
++	tp := &TestPattern{W: 64, H: 48, FPS: 50}
++	ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond)
++	defer cancel()
++	var n, keys int
++	tp.Run(ctx, func(s Sample) {
++		n++
++		if s.Key {
++			keys++
++		}
++	})
++	if n < 3 || keys != 1 {
++		t.Fatalf("frames %d keys %d", n, keys)
++	}
++}
++
++// TestDecodesWithFFmpegLibraries decodes the encoder's output with PyAV (libavcodec) when it is installed and checks
++// the pixels: the strongest evidence the bitstream is valid H.264.
++func TestDecodesWithFFmpegLibraries(t *testing.T) {
++	py, err := exec.LookPath("python3")
++	if err != nil || exec.Command(py, "-c", "import av").Run() != nil {
++		t.Skip("python3 with PyAV (pip install av) not available")
++	}
++	enc, _ := NewPCMEncoder(320, 240)
++	var stream []byte
++	var prev *Frame
++	var last *Frame
++	for i := 0; i < 30; i++ {
++		f := NewFrame(320, 240)
++		DrawPattern(f, i)
++		var coded []int
++		if prev != nil {
++			coded = ChangedMBs(prev, f, 0)
++		}
++		stream = append(stream, enc.Encode(f, i == 20, coded)...)
++		prev, last = f, f
++	}
++	dir := t.TempDir()
++	path := filepath.Join(dir, "p.h264")
++	os.WriteFile(path, stream, 0o600)
++	os.WriteFile(filepath.Join(dir, "y.raw"), last.Y, 0o600)
++	script := `
++import av, sys
++c = av.open(sys.argv[1], format="h264")
++frames = [f for f in c.decode(video=0)]
++last = frames[-1].to_ndarray(format="yuv420p")
++want = open(sys.argv[2], "rb").read()
++y = last[:240].tobytes()
++diff = sum(1 for a, b in zip(y, want) if a != b)
++print(len(frames), frames[-1].width, frames[-1].height, diff)
++`
++	out, err := exec.Command(py, "-c", script, path, filepath.Join(dir, "y.raw")).CombinedOutput()
++	if err != nil {
++		t.Fatalf("decode failed: %v\n%s", err, out)
++	}
++	if got := string(bytes.TrimSpace(out)); got != "30 320 240 0" {
++		t.Fatalf("decoder said %q, want 30 frames 320x240 with the last frame identical", got)
++	}
++}
++
++func TestCommandSource(t *testing.T) {
++	enc, _ := NewPCMEncoder(32, 32)
++	var stream []byte
++	for i := 0; i < 4; i++ {
++		f := NewFrame(32, 32)
++		DrawPattern(f, i)
++		stream = append(stream, enc.Encode(f, i == 0, []int{0, 3})...)
++	}
++	path := filepath.Join(t.TempDir(), "s.h264")
++	os.WriteFile(path, stream, 0o600)
++	cat, err := exec.LookPath("cat")
++	if err != nil {
++		t.Skip("no cat")
++	}
++	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
++	defer cancel()
++	got := make(chan Sample, 100)
++	go (&CommandSource{Argv: []string{cat, path}}).Run(ctx, func(s Sample) { got <- s })
++	for i := 0; i < 4; i++ {
++		select {
++		case s := <-got:
++			if s.Key != (i == 0) {
++				t.Fatalf("sample %d key %v", i, s.Key)
++			}
++		case <-ctx.Done():
++			t.Fatalf("only %d samples", i)
++		}
++	}
++}
++
++func TestJPEGFallback(t *testing.T) {
++	img := image.NewRGBA(image.Rect(0, 0, 64, 48))
++	for i := range img.Pix {
++		img.Pix[i] = byte(i)
++	}
++	var buf bytes.Buffer
++	jpeg.Encode(&buf, img, nil)
++	path := filepath.Join(t.TempDir(), "f.jpg")
++	os.WriteFile(path, buf.Bytes(), 0o600)
++	src := &JPEGSource{FFmpeg: "none", W: 64, H: 48}
++	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
++	defer cancel()
++	got := make(chan Sample, 10)
++	go src.Run(ctx, func(s Sample) { got <- s })
++	src.Push(path)
++	select {
++	case s := <-got:
++		if !s.Key || len(s.Data) < 64*48 {
++			t.Fatalf("key %v len %d", s.Key, len(s.Data))
++		}
++	case <-ctx.Done():
++		t.Fatal("no sample")
++	}
++	src.Push(filepath.Join(t.TempDir(), "missing.jpg")) // ignored
++}
ddiff --git a/internal/video/sources.go b/internal/video/sources.go
nnew file mode 100644
iindex 0000000..750822b
---- /dev/null
++++ b/internal/video/sources.go
@@@ -0,0 +1,494 @@
++package video
++
++import (
++	"bufio"
++	"bytes"
++	"context"
++	"errors"
++	"fmt"
++	"image"
++	"image/jpeg"
++	"io"
++	"log/slog"
++	"os"
++	"os/exec"
++	"sync"
++	"sync/atomic"
++	"time"
++)
++
++// Sample is one H.264 access unit in Annex B form.
++type Sample struct {
++	Data     []byte
++	Key      bool // contains an IDR slice
++	Duration time.Duration
++}
++
++// Source produces samples until ctx ends or it fails. Sink never blocks for long; a source must not wait on it.
++type Source interface {
++	Name() string
++	Run(ctx context.Context, sink func(Sample)) error
++	// RequestKeyframe asks for an IDR as soon as possible (a new viewer session or a PLI).
++	RequestKeyframe()
++}
++
++// ---- test pattern ----
++
++// TestPattern draws colour bars with a moving box and encodes them with the PCM encoder: after the first frame only
++// the macroblocks the box touched are sent.
++type TestPattern struct {
++	W, H, FPS int
++	GOP       time.Duration // keyframe interval (default 4 s)
++	keyReq    atomic.Bool
++}
++
++func (t *TestPattern) Name() string     { return "test-pattern" }
++func (t *TestPattern) RequestKeyframe() { t.keyReq.Store(true) }
++
++var bars = [8][3]byte{ // BT.601 limited range Y, Cb, Cr: white, yellow, cyan, green, magenta, red, blue, black
++	{180, 128, 128}, {162, 44, 142}, {131, 156, 44}, {112, 72, 58}, {84, 184, 198}, {65, 100, 212}, {35, 212, 114}, {16, 128, 128},
++}
++
++// DrawPattern fills f with frame n of the pattern.
++func DrawPattern(f *Frame, n int) {
++	cw, ch := f.W/2, f.H/2
++	for y := 0; y < f.H; y++ {
++		for x := 0; x < f.W; x++ {
++			f.Y[y*f.W+x] = bars[x*8/f.W][0]
++		}
++	}
++	for y := 0; y < ch; y++ {
++		for x := 0; x < cw; x++ {
++			b := bars[x*16/f.W]
++			f.Cb[y*cw+x], f.Cr[y*cw+x] = b[1], b[2]
++		}
++	}
++	// A 32×32 box sweeping left to right across the middle, one macroblock per frame.
++	span := f.W/16 - 1
++	pos := n % (2 * span)
++	if pos >= span {
++		pos = 2*span - pos
++	}
++	bx, by := pos*16, (f.H/2/16)*16-16
++	for y := by; y < by+32 && y < f.H; y++ {
++		for x := bx; x < bx+32 && x < f.W; x++ {
++			f.Y[y*f.W+x] = 235
++		}
++	}
++	for y := by / 2; y < (by+32)/2 && y < ch; y++ {
++		for x := bx / 2; x < (bx+32)/2 && x < cw; x++ {
++			f.Cb[y*cw+x], f.Cr[y*cw+x] = 128, 128
++		}
++	}
++}
++
++func (t *TestPattern) Run(ctx context.Context, sink func(Sample)) error {
++	if t.W == 0 {
++		t.W, t.H = 320, 240
++	}
++	if t.FPS <= 0 {
++		t.FPS = 10
++	}
++	if t.GOP <= 0 {
++		t.GOP = 4 * time.Second
++	}
++	enc, err := NewPCMEncoder(t.W, t.H)
++	if err != nil {
++		return err
++	}
++	period := time.Second / time.Duration(t.FPS)
++	tick := time.NewTicker(period)
++	defer tick.Stop()
++	var prev *Frame
++	lastKey := time.Time{}
++	for n := 0; ; n++ {
++		f := NewFrame(t.W, t.H)
++		DrawPattern(f, n)
++		key := prev == nil || time.Since(lastKey) >= t.GOP || t.keyReq.Swap(false)
++		var coded []int
++		if !key {
++			coded = ChangedMBs(prev, f, 0)
++		} else {
++			lastKey = time.Now()
++		}
++		sink(Sample{Data: enc.Encode(f, key, coded), Key: key, Duration: period})
++		prev = f
++		select {
++		case <-ctx.Done():
++			return nil
++		case <-tick.C:
++		}
++	}
++}
++
++// ---- Annex B parsing ----
++
++// AccessUnits splits an Annex B byte stream into access units, calling emit for each. A new access unit starts at an
++// access unit delimiter, SPS, PPS or SEI after a slice, or at a slice whose first_mb_in_slice is 0.
++func AccessUnits(r io.Reader, emit func(au []byte, key bool)) error {
++	br := bufio.NewReaderSize(r, 1<<20)
++	var au []byte
++	hasVCL, key := false, false
++	flush := func() {
++		if hasVCL {
++			emit(au, key)
++		}
++		au, hasVCL, key = nil, false, false
++	}
++	return splitNALs(br, func(n []byte) {
++		if len(n) == 0 {
++			return
++		}
++		typ := n[0] & 0x1f
++		vcl := typ == 1 || typ == 5
++		if hasVCL && (typ == 9 || typ == 7 || typ == 8 || typ == 6 || (vcl && firstMBZero(n))) {
++			flush()
++		}
++		au = append(au, 0, 0, 0, 1)
++		au = append(au, n...)
++		if vcl {
++			hasVCL = true
++		}
++		if typ == 5 {
++			key = true
++		}
++	}, flush)
++}
++
++// firstMBZero reports whether a slice NAL's first_mb_in_slice (the first ue(v) after the header) is 0, i.e. its
++// first bit is 1.
++func firstMBZero(n []byte) bool { return len(n) > 1 && n[1]&0x80 != 0 }
++
++// splitNALs finds start codes (00 00 01 or 00 00 00 01) and calls nalFn with each NAL (header onwards).
++func splitNALs(br *bufio.Reader, nalFn func([]byte), end func()) error {
++	var cur []byte
++	started := false
++	zeros := 0
++	buf := make([]byte, 64*1024)
++	for {
++		n, err := br.Read(buf)
++		for _, b := range buf[:n] {
++			if b == 0 {
++				zeros++
++				continue
++			}
++			if b == 1 && zeros >= 2 {
++				if started {
++					nalFn(cur)
++				}
++				cur, started, zeros = cur[:0:0], true, 0
++				continue
++			}
++			for ; zeros > 0; zeros-- {
++				cur = append(cur, 0)
++			}
++			cur = append(cur, b)
++		}
++		if err != nil {
++			if started {
++				nalFn(cur)
++			}
++			end()
++			if errors.Is(err, io.EOF) {
++				return nil
++			}
++			return err
++		}
++	}
++}
++
++// ---- a child process writing H.264 ----
++
++// CommandSource runs a program that writes an H.264 Annex B stream to stdout (rpicam-vid, ffmpeg) and restarts it
++// with backoff when it exits.
++type CommandSource struct {
++	Argv []string
++	Log  *slog.Logger
++}
++
++func (c *CommandSource) Name() string { return "command:" + c.Argv[0] }
++
++// RequestKeyframe cannot reach the child; rpicam-vid and ffmpeg are started with a short keyframe interval instead.
++func (c *CommandSource) RequestKeyframe() {}
++
++func (c *CommandSource) Run(ctx context.Context, sink func(Sample)) error {
++	backoff := time.Second
++	for ctx.Err() == nil {
++		start := time.Now()
++		err := c.runOnce(ctx, sink)
++		if ctx.Err() != nil {
++			return nil
++		}
++		if time.Since(start) > time.Minute {
++			backoff = time.Second
++		}
++		if c.Log != nil {
++			c.Log.Warn("video source exited; restarting", "argv0", c.Argv[0], "err", err, "in", backoff)
++		}
++		select {
++		case <-ctx.Done():
++			return nil
++		case <-time.After(backoff):
++		}
++		if backoff < 30*time.Second {
++			backoff *= 2
++		}
++	}
++	return nil
++}
++
++func (c *CommandSource) runOnce(ctx context.Context, sink func(Sample)) error {
++	cmd := exec.CommandContext(ctx, c.Argv[0], c.Argv[1:]...)
++	var stderr bytes.Buffer
++	cmd.Stderr = &limitedWriter{w: &stderr, n: 4096}
++	out, err := cmd.StdoutPipe()
++	if err != nil {
++		return err
++	}
++	if err := cmd.Start(); err != nil {
++		return err
++	}
++	last := time.Now()
++	perr := AccessUnits(out, func(au []byte, key bool) {
++		now := time.Now()
++		d := now.Sub(last)
++		last = now
++		sink(Sample{Data: au, Key: key, Duration: clampDur(d)})
++	})
++	werr := cmd.Wait()
++	if perr != nil {
++		return perr
++	}
++	if werr != nil {
++		return fmt.Errorf("%w: %s", werr, bytes.TrimSpace(stderr.Bytes()))
++	}
++	return errors.New("exited")
++}
++
++func clampDur(d time.Duration) time.Duration {
++	if d < time.Millisecond {
++		return time.Millisecond
++	}
++	if d > time.Second {
++		return time.Second
++	}
++	return d
++}
++
++type limitedWriter struct {
++	w io.Writer
++	n int
++}
++
++func (l *limitedWriter) Write(p []byte) (int, error) {
++	if l.n <= 0 {
++		return len(p), nil
++	}
++	q := p
++	if len(q) > l.n {
++		q = q[:l.n]
++	}
++	l.n -= len(q)
++	_, _ = l.w.Write(q)
++	return len(p), nil
++}
++
++// ---- JPEG frames from a plugin ----
++
++// JPEGSource turns JPEG files announced by a plugin into H.264: through ffmpeg when it is installed, otherwise with
++// the built-in PCM encoder at a low frame rate, sending only the macroblocks that changed.
++type JPEGSource struct {
++	FFmpeg string // path to ffmpeg; empty = look on PATH; "none" = always use the fallback
++	FPS    int
++	W, H   int // output size for the fallback (default 320×240)
++	Log    *slog.Logger
++
++	mu     sync.Mutex
++	latest chan []byte
++	keyReq atomic.Bool
++}
++
++func (j *JPEGSource) Name() string     { return "plugin-jpeg" }
++func (j *JPEGSource) RequestKeyframe() { j.keyReq.Store(true) }
++
++func (j *JPEGSource) ch() chan []byte {
++	j.mu.Lock()
++	defer j.mu.Unlock()
++	if j.latest == nil {
++		j.latest = make(chan []byte, 1)
++	}
++	return j.latest
++}
++
++// Push hands over a frame file. It reads the file at once (plugins rotate their files) and keeps only the newest
++// frame if the encoder is busy.
++func (j *JPEGSource) Push(path string) {
++	b, err := os.ReadFile(path)
++	if err != nil || len(b) < 4 || b[0] != 0xff || b[1] != 0xd8 {
++		return
++	}
++	c := j.ch()
++	select {
++	case c <- b:
++	default:
++		select {
++		case <-c:
++		default:
++		}
++		select {
++		case c <- b:
++		default:
++		}
++	}
++}
++
++func (j *JPEGSource) Run(ctx context.Context, sink func(Sample)) error {
++	if j.FPS <= 0 {
++		j.FPS = 10
++	}
++	ff := j.FFmpeg
++	if ff == "" {
++		ff, _ = exec.LookPath("ffmpeg")
++	} else if ff == "none" {
++		ff = ""
++	}
++	if ff != "" {
++		return j.runFFmpeg(ctx, ff, sink)
++	}
++	if j.Log != nil {
++		j.Log.Warn("ffmpeg not found; publishing camera frames with the built-in low-rate encoder")
++	}
++	return j.runPCM(ctx, sink)
++}
++
++func (j *JPEGSource) runFFmpeg(ctx context.Context, ff string, sink func(Sample)) error {
++	args := []string{"-hide_banner", "-loglevel", "error", "-f", "image2pipe", "-framerate", fmt.Sprint(j.FPS),
++		"-c:v", "mjpeg", "-i", "pipe:0", "-an", "-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency",
++		"-profile:v", "baseline", "-pix_fmt", "yuv420p", "-g", fmt.Sprint(j.FPS * 2), "-bf", "0",
++		"-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2", "-f", "h264", "pipe:1"}
++	cmd := exec.CommandContext(ctx, ff, args...)
++	stdin, err := cmd.StdinPipe()
++	if err != nil {
++		return err
++	}
++	stdout, err := cmd.StdoutPipe()
++	if err != nil {
++		return err
++	}
++	if err := cmd.Start(); err != nil {
++		return err
++	}
++	go func() {
++		defer stdin.Close()
++		c := j.ch()
++		for {
++			select {
++			case <-ctx.Done():
++				return
++			case b := <-c:
++				if _, err := stdin.Write(b); err != nil {
++					return
++				}
++			}
++		}
++	}()
++	last := time.Now()
++	err = AccessUnits(stdout, func(au []byte, key bool) {
++		now := time.Now()
++		sink(Sample{Data: au, Key: key, Duration: clampDur(now.Sub(last))})
++		last = now
++	})
++	_ = cmd.Wait()
++	return err
++}
++
++func (j *JPEGSource) runPCM(ctx context.Context, sink func(Sample)) error {
++	w, h := j.W, j.H
++	if w == 0 {
++		w, h = 320, 240
++	}
++	enc, err := NewPCMEncoder(w, h)
++	if err != nil {
++		return err
++	}
++	fps := j.FPS
++	if fps > 5 {
++		fps = 5
++	}
++	period := time.Second / time.Duration(fps)
++	var prev *Frame
++	lastKey, lastSent := time.Time{}, time.Now()
++	c := j.ch()
++	for {
++		var b []byte
++		select {
++		case <-ctx.Done():
++			return nil
++		case b = <-c:
++		}
++		img, err := jpeg.Decode(bytes.NewReader(b))
++		if err != nil {
++			continue
++		}
++		f := FrameFromImage(img, w, h)
++		key := prev == nil || time.Since(lastKey) > 10*time.Second || j.keyReq.Swap(false)
++		var coded []int
++		if key {
++			lastKey = time.Now()
++		} else {
++			coded = ChangedMBs(prev, f, 6)
++		}
++		now := time.Now()
++		sink(Sample{Data: enc.Encode(f, key, coded), Key: key, Duration: clampDur(now.Sub(lastSent))})
++		lastSent, prev = now, f
++		// Rate limit.
++		select {
++		case <-ctx.Done():
++			return nil
++		case <-time.After(period):
++		}
++	}
++}
++
++// FrameFromImage scales an image to w×h (nearest neighbour) in YUV 4:2:0.
++func FrameFromImage(img image.Image, w, h int) *Frame {
++	f := NewFrame(w, h)
++	b := img.Bounds()
++	cw := w / 2
++	for y := 0; y < h; y++ {
++		sy := b.Min.Y + y*b.Dy()/h
++		for x := 0; x < w; x++ {
++			sx := b.Min.X + x*b.Dx()/w
++			yy, cb, cr := toYCbCr(img, sx, sy)
++			f.Y[y*w+x] = yy
++			if x%2 == 0 && y%2 == 0 {
++				f.Cb[(y/2)*cw+x/2], f.Cr[(y/2)*cw+x/2] = cb, cr
++			}
++		}
++	}
++	return f
++}
++
++func toYCbCr(img image.Image, x, y int) (byte, byte, byte) {
++	if yc, ok := img.(*image.YCbCr); ok {
++		c := yc.YCbCrAt(x, y)
++		return c.Y, c.Cb, c.Cr
++	}
++	if g, ok := img.(*image.Gray); ok {
++		return g.GrayAt(x, y).Y, 128, 128
++	}
++	r, g, bl, _ := img.At(x, y).RGBA()
++	R, G, B := float64(r>>8), float64(g>>8), float64(bl>>8)
++	Y := 0.299*R + 0.587*G + 0.114*B
++	return clampByte(Y), clampByte(128 - 0.168736*R - 0.331264*G + 0.5*B), clampByte(128 + 0.5*R - 0.418688*G - 0.081312*B)
++}
++
++func clampByte(v float64) byte {
++	if v < 1 {
++		return 1
++	}
++	if v > 254 {
++		return 254
++	}
++	return byte(v + 0.5)
++}
ddiff --git a/internal/video/whip.go b/internal/video/whip.go
nnew file mode 100644
iindex 0000000..3e8913f
---- /dev/null
++++ b/internal/video/whip.go
@@@ -0,0 +1,344 @@
++// Package video publishes the device camera to OpenRe over WHIP (RFC 9725) with pion/webrtc: H.264 from a test
++// pattern, from a child process (rpicam-vid, ffmpeg), or from JPEG frames a plugin writes. Video runs on its own
++// goroutines and drops frames rather than wait, so it never blocks control.
++package video
++
++import (
++	"bytes"
++	"context"
++	"errors"
++	"fmt"
++	"io"
++	"log/slog"
++	"net/http"
++	"net/url"
++	"strings"
++	"sync"
++	"sync/atomic"
++	"time"
++
++	"github.com/pion/interceptor"
++	"github.com/pion/rtcp"
++	"github.com/pion/webrtc/v4"
++	"github.com/pion/webrtc/v4/pkg/media"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
++	"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
++)
++
++// Options configure the publisher.
++type Options struct {
++	WHIPURL    string
++	PublishKey credentials.Secret
++	ICEServers []protocol.ICEServer
++	Log        *slog.Logger
++	HTTPClient *http.Client
++	// IncludeLoopback offers loopback ICE candidates (tests on one machine).
++	IncludeLoopback bool
++	ConnectTimeout  time.Duration
++}
++
++// States reported by Publisher.State.
++const (
++	StateOff        = "off"
++	StateConnecting = "connecting"
++	StateLive       = "live"
++	StateRetrying   = "retrying"
++)
++
++// Publisher keeps one WHIP session up and feeds it samples from a Source.
++type Publisher struct {
++	opt    Options
++	log    *slog.Logger
++	state  atomic.Value
++	frames atomic.Int64
++	lastEr atomic.Value
++
++	mu    sync.Mutex
++	track *webrtc.TrackLocalStaticSample
++	needK bool
++	src   Source
++}
++
++func NewPublisher(opt Options) *Publisher {
++	if opt.Log == nil {
++		opt.Log = slog.Default()
++	}
++	if opt.HTTPClient == nil {
++		opt.HTTPClient = &http.Client{Timeout: 20 * time.Second}
++	}
++	if opt.ConnectTimeout == 0 {
++		opt.ConnectTimeout = 20 * time.Second
++	}
++	p := &Publisher{opt: opt, log: opt.Log.With("component", "video")}
++	p.state.Store(StateOff)
++	p.lastEr.Store("")
++	return p
++}
++
++// State is off, connecting, live or retrying.
++func (p *Publisher) State() string { return p.state.Load().(string) }
++
++// Frames counts samples written to a live session.
++func (p *Publisher) Frames() int64 { return p.frames.Load() }
++
++func (p *Publisher) LastError() string { return p.lastEr.Load().(string) }
++
++// Run publishes src until ctx ends, reconnecting with backoff.
++func (p *Publisher) Run(ctx context.Context, src Source) {
++	p.mu.Lock()
++	p.src = src
++	p.mu.Unlock()
++	srcCtx, cancelSrc := context.WithCancel(ctx)
++	defer cancelSrc()
++	go func() {
++		if err := src.Run(srcCtx, p.write); err != nil && srcCtx.Err() == nil {
++			p.log.Error("video source failed", "source", src.Name(), "err", err)
++			p.lastEr.Store("source: " + err.Error())
++		}
++	}()
++	backoff := time.Second
++	for ctx.Err() == nil {
++		start := time.Now()
++		err := p.session(ctx)
++		if ctx.Err() != nil {
++			break
++		}
++		msg := "ended"
++		if err != nil {
++			msg = scrub(err.Error(), p.opt.PublishKey)
++		}
++		p.lastEr.Store(msg)
++		p.state.Store(StateRetrying)
++		if time.Since(start) > time.Minute {
++			backoff = time.Second
++		}
++		p.log.Warn("video session down; retrying", "err", msg, "in", backoff)
++		select {
++		case <-ctx.Done():
++		case <-time.After(backoff):
++		}
++		if backoff < 30*time.Second {
++			backoff *= 2
++		}
++	}
++	p.state.Store(StateOff)
++}
++
++// write is the source's sink: it never blocks on the network (pion queues packets and drops on congestion).
++func (p *Publisher) write(s Sample) {
++	p.mu.Lock()
++	track, needK := p.track, p.needK
++	if track != nil && needK && s.Key {
++		p.needK = false
++		needK = false
++	}
++	p.mu.Unlock()
++	if track == nil || needK {
++		return
++	}
++	if err := track.WriteSample(media.Sample{Data: s.Data, Duration: s.Duration}); err == nil {
++		p.frames.Add(1)
++	}
++}
++
++func (p *Publisher) requestKeyframe() {
++	p.mu.Lock()
++	src := p.src
++	p.mu.Unlock()
++	if src != nil {
++		src.RequestKeyframe()
++	}
++}
++
++func (p *Publisher) newAPI() (*webrtc.API, error) {
++	m := &webrtc.MediaEngine{}
++	if err := m.RegisterCodec(webrtc.RTPCodecParameters{
++		RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000,
++			SDPFmtpLine:  "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f",
++			RTCPFeedback: []webrtc.RTCPFeedback{{Type: "nack"}, {Type: "nack", Parameter: "pli"}, {Type: "ccm", Parameter: "fir"}}},
++		PayloadType: 102,
++	}, webrtc.RTPCodecTypeVideo); err != nil {
++		return nil, err
++	}
++	ir := &interceptor.Registry{}
++	if err := webrtc.RegisterDefaultInterceptors(m, ir); err != nil {
++		return nil, err
++	}
++	se := webrtc.SettingEngine{}
++	if p.opt.IncludeLoopback {
++		se.SetIncludeLoopbackCandidate(true)
++	}
++	return webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithInterceptorRegistry(ir), webrtc.WithSettingEngine(se)), nil
++}
++
++func (p *Publisher) session(ctx context.Context) error {
++	p.state.Store(StateConnecting)
++	api, err := p.newAPI()
++	if err != nil {
++		return err
++	}
++	var ice []webrtc.ICEServer
++	for _, s := range p.opt.ICEServers {
++		ice = append(ice, webrtc.ICEServer{URLs: s.URLs, Username: s.Username, Credential: s.Credential})
++	}
++	pc, err := api.NewPeerConnection(webrtc.Configuration{ICEServers: ice})
++	if err != nil {
++		return err
++	}
++	defer pc.Close()
++	track, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000,
++		SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f"}, "video", "openvibe-node")
++	if err != nil {
++		return err
++	}
++	tr, err := pc.AddTransceiverFromTrack(track, webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly})
++	if err != nil {
++		return err
++	}
++	go func() { // RTCP: a viewer's PLI or FIR asks for a keyframe.
++		for {
++			pkts, _, err := tr.Sender().ReadRTCP()
++			if err != nil {
++				return
++			}
++			for _, pk := range pkts {
++				switch pk.(type) {
++				case *rtcp.PictureLossIndication, *rtcp.FullIntraRequest:
++					p.requestKeyframe()
++				}
++			}
++		}
++	}()
++	state := make(chan webrtc.PeerConnectionState, 8)
++	pc.OnConnectionStateChange(func(s webrtc.PeerConnectionState) {
++		select {
++		case state <- s:
++		default:
++		}
++	})
++	offer, err := pc.CreateOffer(nil)
++	if err != nil {
++		return err
++	}
++	gathered := webrtc.GatheringCompletePromise(pc)
++	if err := pc.SetLocalDescription(offer); err != nil {
++		return err
++	}
++	select {
++	case <-gathered:
++	case <-time.After(5 * time.Second):
++	case <-ctx.Done():
++		return nil
++	}
++	answer, resource, err := p.post(ctx, pc.LocalDescription().SDP)
++	if err != nil {
++		return err
++	}
++	defer p.delete(resource)
++	if err := pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeAnswer, SDP: answer}); err != nil {
++		return fmt.Errorf("whip answer: %w", err)
++	}
++	connectBy := time.NewTimer(p.opt.ConnectTimeout)
++	defer connectBy.Stop()
++	var disconnectedAt time.Time
++	check := time.NewTicker(time.Second)
++	defer check.Stop()
++	defer func() {
++		p.mu.Lock()
++		p.track = nil
++		p.mu.Unlock()
++	}()
++	for {
++		select {
++		case <-ctx.Done():
++			return nil
++		case <-connectBy.C:
++			if p.State() != StateLive {
++				return errors.New("whip: media did not connect in time")
++			}
++		case s := <-state:
++			switch s {
++			case webrtc.PeerConnectionStateConnected:
++				disconnectedAt = time.Time{}
++				p.mu.Lock()
++				p.track, p.needK = track, true
++				p.mu.Unlock()
++				p.requestKeyframe()
++				p.state.Store(StateLive)
++				p.log.Info("video live", "whip", redactURL(p.opt.WHIPURL))
++			case webrtc.PeerConnectionStateDisconnected:
++				disconnectedAt = time.Now()
++			case webrtc.PeerConnectionStateFailed, webrtc.PeerConnectionStateClosed:
++				return fmt.Errorf("whip: connection %s", s)
++			}
++		case <-check.C:
++			if !disconnectedAt.IsZero() && time.Since(disconnectedAt) > 5*time.Second {
++				return errors.New("whip: disconnected")
++			}
++		}
++	}
++}
++
++func (p *Publisher) post(ctx context.Context, sdp string) (answer, resource string, err error) {
++	req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.opt.WHIPURL, strings.NewReader(sdp))
++	if err != nil {
++		return "", "", err
++	}
++	req.Header.Set("Content-Type", "application/sdp")
++	if !p.opt.PublishKey.IsZero() {
++		req.Header.Set("Authorization", "Bearer "+p.opt.PublishKey.Reveal())
++	}
++	resp, err := p.opt.HTTPClient.Do(req)
++	if err != nil {
++		return "", "", fmt.Errorf("whip: %s", scrub(err.Error(), p.opt.PublishKey))
++	}
++	defer resp.Body.Close()
++	body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
++	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK {
++		return "", "", fmt.Errorf("whip: HTTP %d %s", resp.StatusCode, bytes.TrimSpace(body[:min(len(body), 200)]))
++	}
++	loc := resp.Header.Get("Location")
++	if loc != "" {
++		if base, err := url.Parse(p.opt.WHIPURL); err == nil {
++			if u, err := base.Parse(loc); err == nil {
++				loc = u.String()
++			}
++		}
++	}
++	return string(body), loc, nil
++}
++
++func (p *Publisher) delete(resource string) {
++	if resource == "" {
++		return
++	}
++	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
++	defer cancel()
++	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, resource, nil)
++	if err != nil {
++		return
++	}
++	if !p.opt.PublishKey.IsZero() {
++		req.Header.Set("Authorization", "Bearer "+p.opt.PublishKey.Reveal())
++	}
++	if resp, err := p.opt.HTTPClient.Do(req); err == nil {
++		resp.Body.Close()
++	}
++}
++
++func redactURL(s string) string {
++	u, err := url.Parse(s)
++	if err != nil {
++		return "(invalid url)"
++	}
++	u.User, u.RawQuery = nil, ""
++	return u.String()
++}
++
++func scrub(s string, secret credentials.Secret) string {
++	if v := secret.Reveal(); v != "" {
++		s = strings.ReplaceAll(s, v, "[redacted]")
++	}
++	return s
++}
ddiff --git a/internal/video/whip_test.go b/internal/video/whip_test.go
nnew file mode 100644
iindex 0000000..d7396b0
---- /dev/null
++++ b/internal/video/whip_test.go
@@@ -0,0 +1,74 @@
++package video
++
++import (
++	"context"
++	"io"
++	"log/slog"
++	"net/http/httptest"
++	"testing"
++	"time"
++
++	"github.com/OpenVibers/OpenVibe.Node/internal/credentials"
++)
++
++func TestPublishTestPatternOverWHIP(t *testing.T) {
++	rx := &WHIPReceiver{PublishKey: "pk_test"}
++	srv := httptest.NewServer(rx)
++	defer srv.Close()
++	defer rx.Close()
++	p := NewPublisher(Options{WHIPURL: srv.URL + "/whip/dev_1", PublishKey: credentials.NewSecret("pk_test"),
++		IncludeLoopback: true, Log: slog.New(slog.NewTextHandler(io.Discard, nil))})
++	ctx, cancel := context.WithCancel(context.Background())
++	done := make(chan struct{})
++	go func() { p.Run(ctx, &TestPattern{W: 160, H: 96, FPS: 20}); close(done) }()
++	deadline := time.Now().Add(15 * time.Second)
++	for rx.Packets.Load() < 20 {
++		if time.Now().After(deadline) {
++			t.Fatalf("no media: state %s frames %d packets %d err %q", p.State(), p.Frames(), rx.Packets.Load(), p.LastError())
++		}
++		time.Sleep(20 * time.Millisecond)
++	}
++	if p.State() != StateLive {
++		t.Fatalf("state %s", p.State())
++	}
++	cancel()
++	<-done
++	if rx.Deletes.Load() != 1 {
++		t.Fatalf("session not deleted: %d", rx.Deletes.Load())
++	}
++	if rx.BadAuth.Load() != 0 {
++		t.Fatal("publish key not sent")
++	}
++}
++
++func TestPublishRetriesOnBadKey(t *testing.T) {
++	rx := &WHIPReceiver{PublishKey: "right"}
++	srv := httptest.NewServer(rx)
++	defer srv.Close()
++	p := NewPublisher(Options{WHIPURL: srv.URL + "/whip/x", PublishKey: credentials.NewSecret("wrong-key-value"),
++		IncludeLoopback: true, Log: slog.New(slog.NewTextHandler(io.Discard, nil))})
++	ctx, cancel := context.WithCancel(context.Background())
++	done := make(chan struct{})
++	go func() { p.Run(ctx, &TestPattern{W: 32, H: 32, FPS: 5}); close(done) }()
++	deadline := time.Now().Add(5 * time.Second)
++	for p.State() != StateRetrying {
++		if time.Now().After(deadline) {
++			t.Fatalf("state %s", p.State())
++		}
++		time.Sleep(10 * time.Millisecond)
++	}
++	cancel()
++	<-done
++	if e := p.LastError(); e == "" || contains(e, "wrong-key-value") {
++		t.Fatalf("last error %q", e)
++	}
++}
++
++func contains(s, sub string) bool {
++	for i := 0; i+len(sub) <= len(s); i++ {
++		if s[i:i+len(sub)] == sub {
++			return true
++		}
++	}
++	return false
++}
ddiff --git a/internal/video/whiptest.go b/internal/video/whiptest.go
nnew file mode 100644
iindex 0000000..a437aa9
---- /dev/null
++++ b/internal/video/whiptest.go
@@@ -0,0 +1,114 @@
++package video
++
++import (
++	"io"
++	"net/http"
++	"strings"
++	"sync"
++	"sync/atomic"
++	"time"
++
++	"github.com/pion/webrtc/v4"
++)
++
++// WHIPReceiver is a minimal WHIP endpoint for tests (and the fake Bot server): it answers offers, receives the H.264
++// track and counts RTP packets. It checks the bearer publish key.
++type WHIPReceiver struct {
++	PublishKey string
++
++	Packets atomic.Int64
++	Deletes atomic.Int64
++	BadAuth atomic.Int64
++
++	mu  sync.Mutex
++	pcs map[string]*webrtc.PeerConnection
++	n   int
++}
++
++func (w *WHIPReceiver) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
++	if r.Header.Get("Authorization") != "Bearer "+w.PublishKey {
++		w.BadAuth.Add(1)
++		http.Error(rw, "unauthorized", http.StatusUnauthorized)
++		return
++	}
++	switch r.Method {
++	case http.MethodPost:
++		w.post(rw, r)
++	case http.MethodDelete:
++		w.Deletes.Add(1)
++		w.mu.Lock()
++		if pc := w.pcs[r.URL.Path]; pc != nil {
++			pc.Close()
++			delete(w.pcs, r.URL.Path)
++		}
++		w.mu.Unlock()
++		rw.WriteHeader(http.StatusOK)
++	default:
++		http.Error(rw, "method", http.StatusMethodNotAllowed)
++	}
++}
++
++func (w *WHIPReceiver) post(rw http.ResponseWriter, r *http.Request) {
++	if !strings.HasPrefix(r.Header.Get("Content-Type"), "application/sdp") {
++		http.Error(rw, "content type", http.StatusUnsupportedMediaType)
++		return
++	}
++	offer, _ := io.ReadAll(r.Body)
++	m := &webrtc.MediaEngine{}
++	_ = m.RegisterDefaultCodecs()
++	se := webrtc.SettingEngine{}
++	se.SetIncludeLoopbackCandidate(true)
++	api := webrtc.NewAPI(webrtc.WithMediaEngine(m), webrtc.WithSettingEngine(se))
++	pc, err := api.NewPeerConnection(webrtc.Configuration{})
++	if err != nil {
++		http.Error(rw, err.Error(), http.StatusInternalServerError)
++		return
++	}
++	pc.OnTrack(func(t *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
++		for {
++			if _, _, err := t.ReadRTP(); err != nil {
++				return
++			}
++			w.Packets.Add(1)
++		}
++	})
++	if err := pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(offer)}); err != nil {
++		pc.Close()
++		http.Error(rw, err.Error(), http.StatusBadRequest)
++		return
++	}
++	ans, err := pc.CreateAnswer(nil)
++	if err != nil {
++		pc.Close()
++		http.Error(rw, err.Error(), http.StatusBadRequest)
++		return
++	}
++	g := webrtc.GatheringCompletePromise(pc)
++	_ = pc.SetLocalDescription(ans)
++	select {
++	case <-g:
++	case <-time.After(5 * time.Second):
++	}
++	w.mu.Lock()
++	if w.pcs == nil {
++		w.pcs = map[string]*webrtc.PeerConnection{}
++	}
++	w.n++
++	loc := r.URL.Path + "/session" + string(rune('a'+w.n%26))
++	w.pcs[loc] = pc
++	w.mu.Unlock()
++	rw.Header().Set("Content-Type", "application/sdp")
++	rw.Header().Set("Location", loc)
++	rw.WriteHeader(http.StatusCreated)
++	_, _ = rw.Write([]byte(pc.LocalDescription().SDP))
++}
++
++// Close ends every session.
++func (w *WHIPReceiver) Close() {
++	w.mu.Lock()
++	defer w.mu.Unlock()
++	for k, pc := range w.pcs {
++		pc.Close()
++		delete(w.pcs, k)
++	}
++}