feat: per-frame framing on video uni stream (agent + web client)

AGENTS.md

@@ -37,6 +37,18 @@ JSON control messages (bidirectional stream):
 {"type":"fingerprint-refresh","algorithm":"sha-256","fingerprint":"<hex>"}  // sent on connect + cert rotation
 ```
 
+Video uni-stream records use one record per access unit:
+
+| Bytes | Description |
+|-------|-------------|
+| 1 | Flags; bit 0 is set when the access unit contains an IDR NAL |
+| 8 | Timestamp in milliseconds since the stream started, big-endian uint64 |
+| 4 | Access-unit payload length, big-endian uint32 |
+| N | Annex B access unit payload, including start codes |
+
+The client may receive records split across reads and must parse each complete
+record before decoding it.
+
 `start` also accepts optional `codec` and `bitrate`.
 
 On connect the agent pushes `fingerprint-refresh` (only when it manages its own
@@ -75,7 +87,7 @@ await transport.ready;
 const stream = await transport.createBidirectionalStream();
 ```
 
-**Note**: MoQ integration has been removed. The agent now publishes video exclusively over WebTransport unidirectional streams (raw H.264 Annex B).
+**Note**: MoQ integration has been removed. The agent now publishes video exclusively over WebTransport unidirectional streams using the framed H.264 format above.
 
 ### Web UI
 
@@ -112,7 +124,7 @@ backend.Backend { ListDisplays(ctx) ([]Display,error); StartStream(ctx, StartReq
 
 activeBackend (chosen via --backend at startup):
   └─ StartStream → Stream.Chunks() channel
-       └─ publishStream goroutine writes each chunk to every subscriber's WT uni stream
+       └─ publishStream goroutine frames complete access units and writes records to every subscriber's WT uni stream
 ```
 
 ### Start sequence
@@ -121,7 +133,7 @@ activeBackend (chosen via --backend at startup):
 3. Captured returns media socket path → agent connects → reads first frame (header + data) for dimensions
 4. Agent spawns `ffmpeg` with correct `-s WxH`, writes first frame, starts BGRA reader goroutine for subsequent frames
 5. Agent opens unidirectional stream on the caller's session → adds caller as owner + subscriber
-6. ffmpeg stdout read in 64KB chunks → each chunk written to all subscriber unidirectional streams
+6. ffmpeg stdout read in 64KB chunks → chunks pass through the framer, and complete framed access-unit records are written to all subscriber unidirectional streams
 7. Agent responds to caller with `{"type":"started","width":...,"height":...,"codec":"h264"}`
 
 ### Stop sequence

src/framer.go

@@ -0,0 +1,159 @@
+package main
+
+import "encoding/binary"
+
+const maxNALBytes = 8 << 20
+
+type framer struct {
+	buf     []byte
+	pending [][]byte
+	ready   [][]byte
+	dropped int
+}
+
+func (f *framer) Push(b []byte) [][]byte {
+	f.buf = append(f.buf, b...)
+	for {
+		nal, ok := f.nextNAL()
+		if !ok {
+			if len(f.buf) > maxNALBytes {
+				f.buf = nil
+				f.pending = nil
+				f.dropped++
+			}
+			break
+		}
+		f.ingestNAL(nal)
+	}
+	out := f.ready
+	f.ready = nil
+	return out
+}
+
+func (f *framer) Flush() [][]byte {
+	if start := findStartCode(f.buf, 0); start >= 0 {
+		dataStart := start + startCodeLen(f.buf, start)
+		if dataStart < len(f.buf) {
+			f.ingestNAL(append([]byte(nil), f.buf[start:]...))
+		}
+	}
+	f.buf = nil
+	if len(f.pending) > 0 {
+		f.ready = append(f.ready, joinNALs(f.pending))
+		f.pending = nil
+	}
+	out := f.ready
+	f.ready = nil
+	return out
+}
+
+func (f *framer) nextNAL() ([]byte, bool) {
+	start := findStartCode(f.buf, 0)
+	if start < 0 {
+		if len(f.buf) > 3 {
+			f.buf = append([]byte(nil), f.buf[len(f.buf)-3:]...)
+		}
+		return nil, false
+	}
+	dataStart := start + startCodeLen(f.buf, start)
+	next := findStartCode(f.buf, dataStart)
+	if next < 0 {
+		f.buf = append([]byte(nil), f.buf[start:]...)
+		return nil, false
+	}
+	nal := append([]byte(nil), f.buf[start:next]...)
+	f.buf = f.buf[next:]
+	return nal, true
+}
+
+func (f *framer) ingestNAL(nal []byte) {
+	if len(nal) == 0 {
+		return
+	}
+	t := nalType(nal)
+	if isVCL(t) && startsNewPicture(nal) && len(f.pending) > 0 {
+		split := len(f.pending)
+		for split > 0 && !isVCL(nalType(f.pending[split-1])) {
+			split--
+		}
+		if split > 0 {
+			f.ready = append(f.ready, joinNALs(f.pending[:split]))
+			f.pending = append([][]byte(nil), f.pending[split:]...)
+		}
+	}
+	f.pending = append(f.pending, nal)
+}
+
+func joinNALs(nals [][]byte) []byte {
+	size := 0
+	for _, nal := range nals {
+		size += len(nal)
+	}
+	out := make([]byte, 0, size)
+	for _, nal := range nals {
+		out = append(out, nal...)
+	}
+	return out
+}
+
+func findStartCode(buf []byte, from int) int {
+	for i := from; i+3 <= len(buf); i++ {
+		if buf[i] != 0 || buf[i+1] != 0 {
+			continue
+		}
+		if buf[i+2] == 1 {
+			return i
+		}
+		if i+3 < len(buf) && buf[i+2] == 0 && buf[i+3] == 1 {
+			return i
+		}
+	}
+	return -1
+}
+
+func startCodeLen(buf []byte, i int) int {
+	if i+3 < len(buf) && buf[i+2] == 0 && buf[i+3] == 1 {
+		return 4
+	}
+	return 3
+}
+
+func nalType(nal []byte) byte {
+	i := startCodeLen(nal, 0)
+	if i >= len(nal) {
+		return 0
+	}
+	return nal[i] & 0x1f
+}
+
+func isVCL(t byte) bool {
+	return t >= 1 && t <= 5
+}
+
+func startsNewPicture(nal []byte) bool {
+	i := startCodeLen(nal, 0)
+	return i+1 < len(nal) && nal[i+1]&0x80 != 0
+}
+
+func keyframe(au []byte) bool {
+	for i := 0; ; {
+		start := findStartCode(au, i)
+		if start < 0 {
+			return false
+		}
+		t := nalType(au[start:])
+		if t == 5 {
+			return true
+		}
+		i = start + startCodeLen(au, start)
+	}
+}
+
+func encodeFrame(flags byte, tsMs uint64, au []byte) []byte {
+	out := make([]byte, 13+len(au))
+	out[0] = flags
+	binary.BigEndian.PutUint64(out[1:9], tsMs)
+	binary.BigEndian.PutUint32(out[9:13], uint32(len(au)))
+	copy(out[13:], au)
+	return out
+}

src/framer_test.go

@@ -0,0 +1,116 @@
+package main
+
+import (
+	"bytes"
+	"testing"
+)
+
+func TestFramerChunking(t *testing.T) {
+	sps := []byte{0, 0, 0, 1, 0x67, 0x42, 0x00}
+	pps := []byte{0, 0, 1, 0x68, 0xce, 0x06}
+	idr := []byte{0, 0, 0, 1, 0x65, 0x88, 0x11}
+	slice1 := []byte{0, 0, 1, 0x41, 0x9a, 0x22}
+	slice2 := []byte{0, 0, 0, 1, 0x41, 0x9a, 0x33}
+	stream := append(append(append(append(append([]byte{}, sps...), pps...), idr...), slice1...), slice2...)
+	want := [][]byte{
+		append(append(append([]byte{}, sps...), pps...), idr...),
+		slice1,
+		slice2,
+	}
+
+	tests := []struct {
+		name   string
+		chunks [][]byte
+	}{
+		{
+			name: "all at once",
+			chunks: [][]byte{
+				stream,
+			},
+		},
+		{
+			name: "one byte at a time",
+			chunks: func() [][]byte {
+				out := make([][]byte, len(stream))
+				for i := range stream {
+					out[i] = stream[i : i+1]
+				}
+				return out
+			}(),
+		},
+		{
+			name: "split start code",
+			chunks: [][]byte{
+				stream[:len(sps)+len(pps)+len(idr)+1],
+				stream[len(sps)+len(pps)+len(idr)+1 : len(sps)+len(pps)+len(idr)+2],
+				stream[len(sps)+len(pps)+len(idr)+2:],
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var f framer
+			var got [][]byte
+			for _, chunk := range tt.chunks {
+				got = append(got, f.Push(chunk)...)
+			}
+			got = append(got, f.Flush()...)
+			if len(got) != len(want) {
+				t.Fatalf("got %d AUs, want %d", len(got), len(want))
+			}
+			for i := range want {
+				if !bytes.Equal(got[i], want[i]) {
+					t.Errorf("AU %d = %x, want %x", i, got[i], want[i])
+				}
+			}
+			if !keyframe(got[0]) {
+				t.Error("AU 0 is not marked as a keyframe")
+			}
+			if keyframe(got[1]) || keyframe(got[2]) {
+				t.Error("non-IDR AUs marked as keyframes")
+			}
+		})
+	}
+}
+
+func TestEncodeFrame(t *testing.T) {
+	au := []byte{0, 0, 1, 0x65, 0x88}
+	got := encodeFrame(1, 0x0102030405060708, au)
+	want := []byte{
+		1,
+		1, 2, 3, 4, 5, 6, 7, 8,
+		0, 0, 0, 5,
+		0, 0, 1, 0x65, 0x88,
+	}
+	if !bytes.Equal(got, want) {
+		t.Fatalf("frame = %x, want %x", got, want)
+	}
+}
+
+func TestFramerDropsOversizedNALAndResyncs(t *testing.T) {
+	var f framer
+	oversized := make([]byte, maxNALBytes+100)
+	copy(oversized, []byte{0, 0, 0, 1})
+	for i := 4; i < len(oversized); i++ {
+		oversized[i] = 0xff
+	}
+	if got := f.Push(oversized); len(got) != 0 {
+		t.Fatalf("oversized NAL emitted %d AUs", len(got))
+	}
+	if f.dropped != 1 {
+		t.Fatalf("dropped = %d, want 1", f.dropped)
+	}
+
+	sps := []byte{0, 0, 0, 1, 0x67, 0x42, 0x00}
+	pps := []byte{0, 0, 1, 0x68, 0xce, 0x06}
+	idr := []byte{0, 0, 0, 1, 0x65, 0x88, 0x11}
+	nextSlice := []byte{0, 0, 1, 0x41, 0x9a, 0x22}
+	followingSlice := []byte{0, 0, 0, 1, 0x41, 0x9a, 0x33}
+	stream := append(append(append(append(append([]byte{}, sps...), pps...), idr...), nextSlice...), followingSlice...)
+	got := f.Push(stream)
+	want := append(append(append([]byte{}, sps...), pps...), idr...)
+	if len(got) != 1 || !bytes.Equal(got[0], want) {
+		t.Fatalf("resynced AUs = %x, want %x", got, [][]byte{want})
+	}
+}